diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 4a6123c4..3a0d9cb0 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -113,6 +113,8 @@ jobs: repository: nextcloud/context_chat path: apps/context_chat persist-credentials: false + # todo: remove later + ref: feat/noid/ocr-support - name: Checkout backend uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 @@ -151,29 +153,44 @@ jobs: ./occ maintenance:install --verbose --database=${{ matrix.databases }} --database-name=nextcloud --database-host=127.0.0.1 --database-port=$PGSQL_PORT --database-user=root --database-pass=rootpassword --admin-user admin --admin-pass password composer run serve & - - name: Enable context_chat, app_api and testing - run: ./occ app:enable -vvv -f context_chat app_api testing + - name: Enable context_chat and app_api + run: ./occ app:enable -vvv -f context_chat app_api - name: Checkout documentation uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4 with: repository: nextcloud/documentation - path: data/admin/files/documentation + path: documentation persist-credentials: false - name: Prepare docs run: | - cd data/admin/files - mv documentation/admin_manual . - cp -R documentation/developer_manual . - cd developer_manual - find . -type f -name "*.rst" -exec bash -c 'mv "$0" "${0%.rst}.md"' {} \; - cd .. - cp -R documentation/developer_manual ./developer_manual2 - cd developer_manual2 - find . -type f -name "*.rst" -exec bash -c 'mv "$0" "${0%.rst}.txt"' {} \; - cd .. - rm -rf documentation + # images and audio test files + cp -r context_chat_backend/tests/res data/admin/files/ + # admin_manual - text files only, rename .rst -> .md + find documentation/admin_manual -type f \( -name '*.rst' -o -name '*.md' -o -name '*.txt' \) \ + | while IFS= read -r src; do + rel="${src#documentation/admin_manual/}" + dest="data/admin/files/admin_manual/${rel%.rst}.md" + mkdir -p "$(dirname "$dest")" + cp "$src" "$dest" + done + # developer_manual - text files only, rename .rst -> .txt + find documentation/developer_manual -type f \( -name '*.rst' -o -name '*.md' -o -name '*.txt' \) \ + | while IFS= read -r src; do + rel="${src#documentation/developer_manual/}" + dest="data/admin/files/developer_manual/${rel%.rst}.txt" + mkdir -p "$(dirname "$dest")" + cp "$src" "$dest" + done + # user_manual - text files only, no rename + find documentation/user_manual -type f \( -name '*.rst' -o -name '*.md' -o -name '*.txt' \) \ + | while IFS= read -r src; do + rel="${src#documentation/user_manual/}" + dest="data/admin/files/user_manual/${rel}" + mkdir -p "$(dirname "$dest")" + cp "$src" "$dest" + done - name: Run files scan run: | @@ -202,12 +219,34 @@ jobs: echo $! > ../pid.txt # Save the process ID (PID) sleep 60 # Wait for the backend to get ready - - name: Register backend + - name: Setup the fake OCR and STT provider + env: + APP_ID: ccb_test_providers + APP_DISPLAY_NAME: CCB Test Providers + APP_VERSION: '1.0.0' + APP_SECRET: '12345' + APP_HOST: '0.0.0.0' + APP_PORT: '10020' + APP_PERSISTENT_STORAGE: persistent_storage + NEXTCLOUD_URL: http://localhost:8080 + run: | + cd context_chat_backend/tests/ex_app + python3 -u lib/main.py >& ../../ocr_stt_exapp_logs & + + - name: Register backend and the OCR/STT provider run: | timeout 10 ./occ app_api:daemon:register --net host manual_install "Manual Install" manual-install http localhost http://localhost:8080 - timeout 120 ./occ app_api:app:register context_chat_backend manual_install --json-info "{\"appid\":\"context_chat_backend\",\"name\":\"Context Chat Backend\",\"daemon_config_name\":\"manual_install\",\"version\":\"${{ fromJson(steps.appinfo.outputs.result).version }}\",\"secret\":\"12345\",\"port\":10034,\"scopes\":[],\"system_app\":0}" --force-scopes --wait-finish + timeout 120 ./occ app_api:app:register ccb_test_providers manual_install --json-info "{\"appid\":\"ccb_test_providers\",\"name\":\"CCB Test Providers\",\"daemon_config_name\":\"manual_install\",\"version\":\"1.0.0\",\"secret\":\"12345\",\"port\":10020}" --wait-finish + timeout 120 ./occ app_api:app:register context_chat_backend manual_install --json-info "{\"appid\":\"context_chat_backend\",\"name\":\"Context Chat Backend\",\"daemon_config_name\":\"manual_install\",\"version\":\"${{ fromJson(steps.appinfo.outputs.result).version }}\",\"secret\":\"12345\",\"port\":10034}" --wait-finish ls -la context_chat_backend/persistent_storage/* + - name: Enable multimodal indexing + run: | + ./occ context_chat:queue-multimodal-files + + - name: Enable the testing app late so it does not register for the OCR/STT first + run: ./occ app:enable testing + - name: Initial memory usage check run: | ps -p $(cat pid.txt) -o pid,cmd,%mem,rss --sort=-%mem @@ -233,7 +272,7 @@ jobs: run: | success=0 echo "::group::Checking stats periodically for 15 minutes to allow the backend to index the files" - for i in {1..90}; do + for i in {1..30}; do echo "Checking stats, attempt $i..." stats_err=$(mktemp) @@ -251,7 +290,7 @@ jobs: # Check for critical errors in output if [ $stats_exit -ne 0 ] || echo "$stats" | grep -q "Error during request"; then echo "Backend connection error detected (exit=$stats_exit), retrying..." - sleep 10 + sleep 30 continue fi @@ -260,9 +299,11 @@ jobs: # Extract indexed documents count (files__default) indexed_count=$(echo "$stats" | jq '.vectordb_document_counts.files__default' || echo "") + queued_count=$(echo "$stats" | jq '.queued_documents_counts.files__default // empty' || echo "") echo "Total eligible files: $total_eligible_files" echo "Indexed documents (files__default): $indexed_count" + echo "Queued documents (files__default): $queued_count" diff=$((total_eligible_files - indexed_count)) threshold=$((total_eligible_files * 3 / 100)) @@ -277,6 +318,12 @@ jobs: echo "Outside 3% tolerance: diff=$diff (${progress}%), threshold=$threshold" fi + # Exit early if no queued documents remain since nothing more will be indexed + if [ "$queued_count" = "0" ]; then + echo "No queued documents remaining but indexing outside tolerance (diff=$diff, threshold=$threshold). Possible locked/failed documents." + break + fi + # Check if backend is still alive ccb_alive=$(ps -p $(cat pid.txt) -o cmd= | grep -c "main.py" || echo "0") if [ "$ccb_alive" -eq 0 ]; then @@ -284,7 +331,7 @@ jobs: exit 1 fi - sleep 10 + sleep 30 done echo "::endgroup::" @@ -299,17 +346,40 @@ jobs: ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' > worker1_logs 2>&1 & ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' > worker2_logs 2>&1 & - echo ::group::English prompt - OUT1=$(./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?") - echo "$OUT1" - echo "$OUT1" | grep -q "If all of these points are met, we give a Green label." || exit 1 - echo ::endgroup:: + # Each entry: "Label;prompt text;grep pattern;grep flags" + checks=( + "Text - Ethical AI Rating (EN);Which factors are taken into account for the Ethical AI Rating?;If all of these points are met, we give a Green label.;" + "Text - Ethical AI Rating (DE);Welche Faktoren beeinflussen das Ethical AI Rating?;If all of these points are met, we give a Green label.;" + "Text - occ maintenance mode;How do I enable maintenance mode in Nextcloud using the command line?;maintenance:mode --on;-iE" + "Text - background job configuration;What are the available background job execution methods in Nextcloud?;cron|webcron|ajax;-iE" + "STT - Amazon oxygen fraction;What fraction of the world's oxygen supply does the Amazon rainforest produce?;twenty percent;-i" + "STT - quantum computing;How does quantum computing perform calculations?;superposition|entanglement;-iE" + "STT - Jungle Book;Who raised Mowgli in the jungle and which villain does he face?;wolves|Shere Khan;-iE" + "OCR - invoice total;What is the total amount on invoice number 10042?;299;-E" + "OCR - train to Amsterdam;What time does the express train to Amsterdam depart from Platform 3?;08:14;" + ) + + failed=0 + for entry in "${checks[@]}"; do + IFS=';' read -r label prompt pattern flags <<< "$entry" + OUT=$(./occ context_chat:prompt admin "$prompt") + # shellcheck disable=SC2086 + if echo "$OUT" | grep -q $flags "$pattern"; then + status="PASS" + else + status="FAIL" + failed=1 + fi + echo "::group::[$status] $label" + echo "$OUT" + if [ "$status" = "FAIL" ]; then + echo "FAIL: expected to find '$pattern' in output" + fi + echo "::endgroup::" + sleep 10 + done - echo ::group::German prompt - OUT2=$(./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?") - echo "$OUT2" - echo "$OUT2" | grep -q "If all of these points are met, we give a Green label." || exit 1 - echo ::endgroup:: + [ $failed -eq 0 ] || exit 1 - name: Check python memory usage run: | @@ -349,6 +419,11 @@ jobs: run: | tail -v -n +1 worker?_logs || echo "No worker logs" + - name: Show ccb_test_providers app logs + if: always() + run: | + cat context_chat_backend/ocr_stt_exapp_logs || echo "No ccb_test_providers app logs" + - name: Show main app logs if: always() run: | diff --git a/REUSE.toml b/REUSE.toml index 48e013fd..06880da4 100644 --- a/REUSE.toml +++ b/REUSE.toml @@ -10,3 +10,9 @@ path = ["requirements.txt", "requirements_dev.txt"] precedence = "aggregate" SPDX-FileCopyrightText = "2024 Nextcloud GmbH and Nextcloud contributors" SPDX-License-Identifier = "AGPL-3.0-or-later" + +[[annotations]] +path = ["tests/res/**"] +precedence = "aggregate" +SPDX-FileCopyrightText = "2026 Nextcloud GmbH and Nextcloud contributors" +SPDX-License-Identifier = "CC0-1.0" diff --git a/context_chat_backend/chain/ingest/doc_loader.py b/context_chat_backend/chain/ingest/doc_loader.py index 832c8331..55422285 100644 --- a/context_chat_backend/chain/ingest/doc_loader.py +++ b/context_chat_backend/chain/ingest/doc_loader.py @@ -3,10 +3,13 @@ # SPDX-License-Identifier: AGPL-3.0-or-later # +import asyncio +import logging import re import tempfile from collections.abc import Callable from io import BytesIO +from sys import getsizeof import docx2txt from epub2txt import epub2txt @@ -17,8 +20,26 @@ from pypdf.errors import FileNotDecryptedError as PdfFileNotDecryptedError from striprtf import striprtf -from ...types import IndexingException, SourceItem - +from ...types import IndexingException, SourceItem, TaskProcClientException, TaskProcException, TaskProcFatalException +from .task_proc import ( + OCR_TASK_TYPE, + SPEECH_TO_TEXT_TASK_TYPE, + RetryableException, + delete_temp_files, + do_ocr, + do_transcription, + is_task_type_available, + upload_temp_files, +) + +logger = logging.getLogger('ccb.doc_loader') +PDF_IMAGES_BATCH_SIZE = 25 +PDF_IMAGES_MAX_SIZE = 100 * 1024 * 1024 # 100 MiB in bytes +# Use callables so the check is deferred to worker process execution time, +# avoiding a stale value inherited from the forkserver preload phase. +# is_task_type_available is cached for 15 minutes so repeated calls are cheap. +is_ocr_available: Callable[[], bool] = lambda: asyncio.run(is_task_type_available(OCR_TASK_TYPE)) +is_stt_available: Callable[[], bool] = lambda: asyncio.run(is_task_type_available(SPEECH_TO_TEXT_TASK_TYPE)) def _temp_file_wrapper(file: BytesIO, loader: Callable, sep: str = '\n') -> str: raw_bytes = file.read() @@ -34,46 +55,99 @@ def _temp_file_wrapper(file: BytesIO, loader: Callable, sep: str = '\n') -> str: # -- LOADERS -- # -def _load_pdf(file: BytesIO) -> str: +def _load_pdf(file: BytesIO, source: SourceItem) -> str: + # ocr_enabled = is_ocr_available() + # todo: skip processsing images inside PDFs for now + ocr_enabled = False pdf_reader = PdfReader(file) - return '\n\n'.join([page.extract_text().strip() for page in pdf_reader.pages]) - - -def _load_csv(file: BytesIO) -> str: + output = [] + + for page in pdf_reader.pages: + text = page.extract_text().strip() + page_ocr_outputs = [] + + if ocr_enabled: + for i in range(0, len(page.images), PDF_IMAGES_BATCH_SIZE): + image_files = {} + for img in page.images[i:i+PDF_IMAGES_BATCH_SIZE]: + if getsizeof(img.data) > PDF_IMAGES_MAX_SIZE: + logger.info( + f'An image {img.name} embedded in a PDF {source.reference}' + f' exceeds max allowed size of {PDF_IMAGES_MAX_SIZE} bytes' + ) + image_files[img.name] = img.data + + try: + file_ids_map = asyncio.run(upload_temp_files(image_files)) + except RetryableException: + raise + except Exception as e: + logger.warning( + f'Error during uploading an embedded PDF image from {source.reference}: {e}', + exc_info=e, + ) + continue + + try: + ocr_tp_outputs = asyncio.run(do_ocr(source.userIds[0], list(file_ids_map.values()))) # pyright: ignore[reportArgumentType] + asyncio.run(delete_temp_files(file_ids_map.values())) # pyright: ignore[reportArgumentType] + except TaskProcFatalException as e: + # task type is not present anymore, stop OCR for this document + logger.warning( + 'The OCR provider disappeared mid-operation, disabling OCR for this batch of' + f' documents including {source.reference}: {e}' + ) + ocr_enabled = False + break + except TaskProcException as e: + logger.warning(f'OCR for embedded images in PDF {source.reference} failed: {e}', exc_info=e) + continue + + # for each image batch + page_ocr_outputs += ocr_tp_outputs + + # for each page, append text then its OCR outputs to preserve document order + output.append(text) + output += page_ocr_outputs + + return '\n\n'.join(output) + + +def _load_csv(file: BytesIO, _: SourceItem) -> str: return read_csv(file).to_string(header=False, na_rep='') -def _load_epub(file: BytesIO) -> str: +def _load_epub(file: BytesIO, _: SourceItem) -> str: return _temp_file_wrapper(file, epub2txt).strip() -def _load_docx(file: BytesIO) -> str: +def _load_docx(file: BytesIO, _: SourceItem) -> str: return docx2txt.process(file).strip() -def _load_odt(file: BytesIO) -> str: +def _load_odt(file: BytesIO, _: SourceItem) -> str: return _temp_file_wrapper(file, lambda fp: Document(fp).get_formatted_text()).strip() -def _load_ppt_x(file: BytesIO) -> str: +def _load_ppt_x(file: BytesIO, _: SourceItem) -> str: return _temp_file_wrapper(file, lambda fp: UnstructuredLoader(fp).load()).strip() -def _load_rtf(file: BytesIO) -> str: +def _load_rtf(file: BytesIO, _: SourceItem) -> str: return striprtf.rtf_to_text(file.read().decode('utf-8', 'ignore')).strip() -def _load_xml(file: BytesIO) -> str: +def _load_xml(file: BytesIO, _: SourceItem) -> str: data = file.read().decode('utf-8', 'ignore') data = re.sub(r'', '', data) return data.strip() -def _load_xlsx(file: BytesIO) -> str: +def _load_xlsx(file: BytesIO, _: SourceItem) -> str: return read_excel(file, na_filter=False).to_string(header=False, na_rep='') -def _load_email(file: BytesIO, ext: str = 'eml') -> str: +def _load_email(file: BytesIO, _: SourceItem, ext: str = 'eml') -> str | None: # NOTE: msg format is not tested if ext not in ['eml', 'msg']: raise IndexingException(f'Unsupported email format: {ext}') @@ -128,13 +202,51 @@ def decode_source(source: SourceItem) -> str: if source.title.endswith('.pot'): raise IndexingException('PowerPoint template files (.pot) are not supported') + try: + if source.type.startswith('image/'): + if is_ocr_available(): + return asyncio.run(do_ocr(source.userIds[0], [source.file_id]))[0] + raise IndexingException( + f'Image file ({source.reference}) cannot be processed since OCR task type is not present', + retryable=True, + ) + if source.type.startswith('audio/'): + if is_stt_available(): + return asyncio.run(do_transcription(source.userIds[0], source.file_id)) + raise IndexingException( + f'Audio file ({source.reference}) cannot be processed since Speech-to-Text task type is not present', # noqa: E501 + retryable=True, + ) + except TaskProcFatalException as e: + logger.warning( + 'The OCR provider disappeared mid-operation, disabling OCR for this batch of' + f' documents including {source.reference}: {e}' + ) + raise IndexingException( # noqa: B904 + f'Image file ({source.reference}) cannot be processed since OCR task type is not present', + retryable=True, + ) + except TaskProcClientException as e: + logger.warning(f'OCR task failed for source file ({source.reference}): {e}') + raise IndexingException(f'OCR task failed for source file ({source.reference}): {e}') # noqa: B904 + except TaskProcException as e: + logger.warning(f'OCR task failed for source file ({source.reference}), it will be retried: {e}') + raise IndexingException( # noqa: B904 + f'OCR task failed for source file ({source.reference}), it will be retried: {e}', + retryable=True, + ) + except ValueError: + # should not happen + logger.warning(f'Unexpected ValueError for source file ({source.reference})') + raise IndexingException(f'Unexpected ValueError for source file ({source.reference})') # noqa: B904 + if isinstance(source.content, str): io_obj = BytesIO(source.content.encode('utf-8', 'ignore')) else: io_obj = source.content if _loader_map.get(source.type): - result = _loader_map[source.type](io_obj) + result = _loader_map[source.type](io_obj, source) return result.encode('utf-8', 'ignore').decode('utf-8', 'ignore').strip() return io_obj.read().decode('utf-8', 'ignore').strip() diff --git a/context_chat_backend/chain/ingest/injest.py b/context_chat_backend/chain/ingest/injest.py index ad2777ed..eb79f9f3 100644 --- a/context_chat_backend/chain/ingest/injest.py +++ b/context_chat_backend/chain/ingest/injest.py @@ -14,18 +14,44 @@ from nc_py_api import AsyncNextcloudApp from ...dyn_loader import VectorDBLoader -from ...types import IndexingError, IndexingException, ReceivedFileItem, SourceItem, TConfig +from ...mimetype_list import AUDIO_MIMETYPES, IMAGE_MIMETYPES, TEXT_MIMETYPES +from ...types import FILES_PROVIDER_ID, IndexingError, IndexingException, ReceivedFileItem, SourceItem, TConfig from ...vectordb.base import BaseVectorDB from ...vectordb.types import DbException, SafeDbException, UpdateAccessOp from ..types import InDocument from .doc_loader import decode_source from .doc_splitter import get_splitter_for +from .task_proc import OCR_TASK_TYPE, SPEECH_TO_TEXT_TASK_TYPE, is_task_type_available logger = logging.getLogger('ccb.injest') MAX_FILE_SIZE = 100 * 1024 * 1024 # 100 MB, all loaded in RAM at once +def _do_extended_mimetype_validation( + source: SourceItem | ReceivedFileItem, + ocr_available: bool, + stt_available: bool, +) -> None: + ''' + Raises + ------ + ValueError + ''' + + extended_mimetypes = ( + *TEXT_MIMETYPES, + *(([], IMAGE_MIMETYPES)[ocr_available]), + *(([], AUDIO_MIMETYPES)[stt_available]), + ) + + if source.reference.startswith(FILES_PROVIDER_ID) and source.type not in extended_mimetypes: + raise ValueError( + f'Unsupported file type: {source.type} for reference {source.reference}.' + f' OCR available: {ocr_available}, Speech-to-text available: {stt_available}.' + ) + + async def __fetch_file_content( semaphore: asyncio.Semaphore, file_id: int, @@ -191,6 +217,7 @@ def _filter_sources( try: existing_source_ids, to_embed_source_ids = vectordb.check_sources(sources) except Exception as e: + # todo: maybe handle this and other errors as IndexingErrors raise DbException('Error: Vectordb error while checking existing sources in indexing') from e existing_sources = {} @@ -215,6 +242,7 @@ def _sources_to_indocuments( for db_id, source in sources.items(): logger.debug('processing source', extra={ 'source_id': source.reference }) + # todo: maybe fetch the content of the files here # transform the source to have text data try: logger.debug('Decoding source %s (type: %s)', source.reference, source.type) @@ -283,7 +311,7 @@ def _sources_to_indocuments( def _increase_access_for_existing_sources( vectordb: BaseVectorDB, existing_sources: Mapping[int, SourceItem | ReceivedFileItem] -) -> Mapping[int, IndexingError | None]: +) -> dict[int, IndexingError | None]: ''' update userIds for existing sources allow the userIds as additional users, not as the only users @@ -324,7 +352,7 @@ def _process_sources( vectordb: BaseVectorDB, config: TConfig, sources: Mapping[int, SourceItem | ReceivedFileItem] -) -> Mapping[int, IndexingError | None]: +) -> dict[int, IndexingError | None]: ''' Processes the sources and adds them to the vectordb. Returns the list of source ids that were successfully added and those that need to be retried. @@ -411,5 +439,24 @@ def embed_sources( 'len(source_ids)': len(sources), }) + mime_filtered_sources: dict[int, SourceItem | ReceivedFileItem] = {} + mime_errored_sources: dict[int, IndexingError] = {} + + ocr_available = asyncio.run(is_task_type_available(OCR_TASK_TYPE)) + stt_available = asyncio.run(is_task_type_available(SPEECH_TO_TEXT_TASK_TYPE)) + for db_id, source in sources.items(): + try: + _do_extended_mimetype_validation(source, ocr_available, stt_available) + mime_filtered_sources[db_id] = source + except ValueError as e: + mime_errored_sources[db_id] = IndexingError( + error=str(e), + retryable=False, + ) + continue + vectordb = vectordb_loader.load() - return _process_sources(vectordb, config, sources) + processed_sources = _process_sources(vectordb, config, mime_filtered_sources) + processed_sources.update(mime_errored_sources) + + return processed_sources diff --git a/context_chat_backend/chain/ingest/task_proc.py b/context_chat_backend/chain/ingest/task_proc.py new file mode 100644 index 00000000..c57ab311 --- /dev/null +++ b/context_chat_backend/chain/ingest/task_proc.py @@ -0,0 +1,356 @@ +# +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# + +import asyncio +import json +import logging +import os +from typing import Any, Literal + +import niquests +from nc_py_api import AsyncNextcloudApp, NextcloudException +from pydantic import BaseModel, ValidationError + +from ...types import TaskProcClientException, TaskProcException, TaskProcFatalException +from ...utils import timed_cache_async + +LOGGER = logging.getLogger('ccb.task_proc') +OCR_TASK_TYPE = 'core:image2text:ocr' +SPEECH_TO_TEXT_TASK_TYPE = 'core:audio2text' +CACHE_TTL = 15 * 60 # cache values for 15 minutes +OCP_TASK_PROC_SCHED_RETRIES = 3 +OCP_TASK_TIMEOUT = 20 * 60 # 20 mins to wait for a task to complete + + +class Task(BaseModel): + id: int + status: str + output: dict[str, Any] | None = None + + +class TaskResponse(BaseModel): + task: Task + + +InputShapeType = Literal[ + 'Number', + 'Text', + 'Audio', + 'Image', + 'Video', + 'File', + 'Enum', + 'ListOfNumbers', + 'ListOfTexts', + 'ListOfImages', + 'ListOfAudios', + 'ListOfVideos', + 'ListOfFiles', +] + +class InputShape(BaseModel): + name: str + description: str + type: InputShapeType + + +class InputShapeEnum(BaseModel): + name: str + value: str + + +class TaskType(BaseModel): + name: str + description: str + inputShape: dict[str, InputShape] + inputShapeEnumValues: dict[str, list[InputShapeEnum]] + inputShapeDefaults: dict[str, str | int | float] + optionalInputShape: dict[str, InputShape] + optionalInputShapeEnumValues: dict[str, list[InputShapeEnum]] + optionalInputShapeDefaults: dict[str, str | int | float] + outputShape: dict[str, InputShape] + outputShapeEnumValues: dict[str, list[InputShapeEnum]] + optionalOutputShape: dict[str, InputShape] + optionalOutputShapeEnumValues: dict[str, list[InputShapeEnum]] + + +class TaskTypesResponse(BaseModel): + types: dict[str, TaskType] + + +class UploadFileError(BaseModel): + message: str + code: int + + +class UploadFileResponse(BaseModel): + fileIds: dict[str, int] + errors: dict[str, UploadFileError] + + +class RetryableException(Exception): + ... + + +def __try_parse_ocs_response(response: niquests.Response | None) -> dict | str: + if response is None or response.text is None: + return 'No response' + try: + ocs_response = json.loads(response.text) + if not (ocs_data := ocs_response.get('ocs', {}).get('data')): + return response.text + return ocs_data + except json.JSONDecodeError: + return response.text + + +async def __schedule_task(user_id: str, task_type: str, custom_id: str, task_input: dict) -> Task: + ''' + Raises + ------ + TaskProcException + ''' + nc = AsyncNextcloudApp() + await nc.set_user(user_id) + + for sched_tries in range(OCP_TASK_PROC_SCHED_RETRIES): + try: + response = await nc.ocs( + 'POST', + '/ocs/v2.php/taskprocessing/schedule', + json={ + 'type': task_type, + 'appId': os.getenv('APP_ID', 'context_chat_backend'), + 'customId': f'ccb-{custom_id}', + 'input': task_input, + }, + ) + try: + task = TaskResponse.model_validate(response).task + LOGGER.debug('TaskProcessing task schedule response', extra={ + 'task': task, + }) + return task + except ValidationError as e: + raise TaskProcException('Failed to parse TaskProcessing task result') from e + except NextcloudException as e: + if e.status_code == niquests.codes.precondition_failed: # type: ignore[attr-defined] + raise TaskProcFatalException( + 'Failed to schedule Nextcloud TaskProcessing task:' + f' No provider of {task_type} is installed on this Nextcloud instance.' + ' Please install a suitable provider from the AI overview:' + ' https://docs.nextcloud.com/server/latest/admin_manual/ai/overview.html.', + ) from e + + if e.status_code == niquests.codes.too_many_requests: # type: ignore[attr-defined] + LOGGER.warning( + 'Rate limited during TaskProcessing task scheduling, waiting 30s before retrying', + extra={ + 'task_type': task_type, + 'sched_try': sched_tries, + }, + ) + await asyncio.sleep(30) + continue + + ocs_response = __try_parse_ocs_response(e.response) + if e.status_code // 100 == 4: + raise TaskProcClientException( + f'Failed to schedule TaskProcessing task due to client error: {ocs_response}', + ) from e + + LOGGER.error('NextcloudException during TaskProcessing task scheduling', exc_info=e, extra={ + 'task_type': task_type, + 'sched_try': sched_tries, + 'nc_exc_reason': str(e.reason), + 'nc_exc_info': str(e.info), + 'nc_exc_status_code': str(e.status_code), + 'ocs_response': str(ocs_response), + }) + raise TaskProcException(f'Failed to schedule TaskProcessing task: {ocs_response}') from e + except TaskProcException: + raise + except Exception as e: + raise TaskProcException(f'Failed to schedule TaskProcessing task: {e}') from e + + raise TaskProcException('Failed to schedule TaskProcessing task, tried 3 times') + + +async def __get_task_result(user_id: str, task: Task) -> Any: + nc = AsyncNextcloudApp() + await nc.set_user(user_id) + + i = 0 + now_waiting_for = 0 + + while task.status != 'STATUS_SUCCESSFUL' and task.status != 'STATUS_FAILED' and now_waiting_for < OCP_TASK_TIMEOUT: + i += 1 + now_waiting_for += 10 + await asyncio.sleep(10) + + try: + response = await nc.ocs('GET', f'/ocs/v2.php/taskprocessing/task/{task.id}') + except NextcloudException as e: + if e.status_code == niquests.codes.too_many_requests: # type: ignore[attr-defined] + LOGGER.warning( + 'Rate limited during TaskProcessing task polling, waiting 10s before retrying', + extra={ + 'task_id': task.id, + 'tries_so_far': i, + 'waiting_time': now_waiting_for, + }, + ) + now_waiting_for += 60 + await asyncio.sleep(60) + continue + raise TaskProcException('Failed to poll TaskProcessing task') from e + except niquests.RequestException as e: + LOGGER.warning('Ignored error during TaskProcessing task polling', exc_info=e, extra={ + 'task_id': task.id, + 'tries_so_far': i, + 'waiting_time': now_waiting_for, + }) + continue + + try: + task = TaskResponse.model_validate(response).task + LOGGER.debug(f'TaskProcessing task poll ({now_waiting_for}s) response', extra={ + 'task_id': task.id, + 'tries_so_far': i, + 'waiting_time': now_waiting_for, + 'task': task, + }) + except ValidationError as e: + raise TaskProcException('Failed to parse TaskProcessing task result') from e + + if task.status != 'STATUS_SUCCESSFUL': + # it's not possible to know if the document is unprocessable or the provider is having a temporary issue + # so we retry it + raise TaskProcException( + f'TaskProcessing task id {task.id} failed with status {task.status}' + f' after waiting {now_waiting_for} seconds', + ) + + if not isinstance(task.output, dict) or 'output' not in task.output: + raise TaskProcException(f'"output" key not found or invalid in TaskProcessing task result: {task.output}') + + return task.output['output'] + + +async def do_ocr(user_id: str, file_ids: list[int]) -> list[str]: + try: + task = await __schedule_task(user_id, OCR_TASK_TYPE, 'ocr', {'input': file_ids}) + output = await __get_task_result(user_id, task) + if not isinstance(output, list) or len(output) == 0 or not isinstance(output[0], str): + raise TaskProcException(f'OCR task returned empty or invalid output: {output}') + return output + except TaskProcException as e: + LOGGER.error(f'Failed to perform OCR for file_ids {file_ids}', exc_info=e) + raise + + +async def do_transcription(user_id: str, file_id: int) -> str: + try: + task = await __schedule_task(user_id, SPEECH_TO_TEXT_TASK_TYPE, str(file_id), {'input': file_id}) + output = await __get_task_result(user_id, task) + if not isinstance(output, str) or len(output.strip()) == 0: + raise TaskProcException(f'Speech-to-text task returned empty or invalid output: {output}') + return output + except TaskProcException as e: + LOGGER.error(f'Failed to perform transcription for file_id {file_id}', exc_info=e) + raise + + +@timed_cache_async(CACHE_TTL) +async def __get_task_types() -> TaskTypesResponse: + ''' + Raises + ------ + TaskProcException + ''' + nc = AsyncNextcloudApp() + + # NC 33 required for this + try: + response = await nc.ocs( + 'GET', + '/ocs/v2.php/taskprocessing/tasks_consumer/tasktypes', + ) + except NextcloudException as e: + raise TaskProcException('Failed to fetch Nextcloud TaskProcessing types') from e + + try: + task_types = TaskTypesResponse.model_validate(response) + LOGGER.debug('Fetched task types', extra={ + 'task_types': task_types, + }) + except (KeyError, TypeError, ValidationError) as e: + raise TaskProcException('Failed to parse Nextcloud TaskProcessing types') from e + + return task_types + + +@timed_cache_async(CACHE_TTL) +async def is_task_type_available(task_type: str) -> bool: + try: + task_types = await __get_task_types() + except Exception as e: + LOGGER.warning(f'Failed to fetch task types: {e}', exc_info=e) + return False + if task_type not in task_types.types: + return False + return True + + +async def upload_temp_files(files: dict[str, str | bytes | bytearray], _tries: int = 1) -> dict[str, int]: + ''' + Returns + ------- + dict[str, int]: NC file ids of the uploaded files in the format {[filename: str]: int} + + Raises + ------ + niquests.RequestException + nc_py_api._exceptions.NextcloudException + pydantic.ValidationError + RetryableException: the upload or the pdf file as a whole should be retried later + ''' + + try: + nc = AsyncNextcloudApp() + res = await nc.ocs( + 'POST', + '/ocs/v2.php/apps/context_chat/upload_files', + files=files, + ) + vald_response: UploadFileResponse = UploadFileResponse.model_validate(res) + if vald_response.errors: + for err in vald_response.errors.values(): + if err.code // 100 == 5: + raise RetryableException(err.message) + + return vald_response.fileIds + except niquests.RequestException as e: + if not e.response or not e.response.status_code: + raise + if e.response.status_code // 100 == 4 and _tries > 0: + return await upload_temp_files(files, _tries - 1) + raise + except NextcloudException as e: + reason = e.response.text if e.response else e.reason + LOGGER.warning('NextcloudException occurred during temp image file upload from PDF: %s', reason) + raise + + +async def delete_temp_files(file_ids: list[int]): + try: + nc = AsyncNextcloudApp() + await nc.ocs( + 'DELETE', + '/ocs/v2.php/apps/context_chat/temp_files', + json={ 'fileIds': file_ids }, + ) + except Exception as e: + LOGGER.warning(f'Unable to delete temporary files from NC appdata: {e}', exc_info=e) diff --git a/context_chat_backend/mimetype_list.py b/context_chat_backend/mimetype_list.py index 87f10241..5dd2da57 100644 --- a/context_chat_backend/mimetype_list.py +++ b/context_chat_backend/mimetype_list.py @@ -2,7 +2,7 @@ # SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors # SPDX-License-Identifier: AGPL-3.0-or-later # -SUPPORTED_MIMETYPES = [ +TEXT_MIMETYPES = ( 'text/plain', 'text/markdown', 'application/json', @@ -22,4 +22,35 @@ 'message/rfc822', 'application/vnd.ms-outlook', 'text/org', -] +) + +IMAGE_MIMETYPES = ( + 'image/bmp', + 'image/bpg', + 'image/emf', + 'image/gif', + 'image/heic', + 'image/heif', + 'image/jp2', + 'image/jpeg', + 'image/png', + 'image/svg+xml', + 'image/tga', + 'image/tiff', + 'image/webp', + 'image/x-dcraw', + 'image/x-icon', +) + +AUDIO_MIMETYPES = ( + 'audio/aac', + 'audio/flac', + 'audio/mp4', + 'audio/mpeg', + 'audio/ogg', + 'audio/wav', + 'audio/webm', + 'audio/x-scpls', +) + +SUPPORTED_MIMETYPES = TEXT_MIMETYPES + IMAGE_MIMETYPES + AUDIO_MIMETYPES diff --git a/context_chat_backend/types.py b/context_chat_backend/types.py index 700d7ddf..664fe4dd 100644 --- a/context_chat_backend/types.py +++ b/context_chat_backend/types.py @@ -14,16 +14,6 @@ from .mimetype_list import SUPPORTED_MIMETYPES from .vectordb.types import UpdateAccessOp -__all__ = [ - 'DEFAULT_EM_MODEL_ALIAS', - 'EmbeddingException', - 'LoaderException', - 'TConfig', - 'TEmbeddingAuthApiKey', - 'TEmbeddingAuthBasic', - 'TEmbeddingConfig', -] - DEFAULT_EM_MODEL_ALIAS = 'em_model' FILES_PROVIDER_ID = 'files__default' @@ -152,6 +142,18 @@ class DocErrorEmbeddingException(EmbeddingException): """ +class TaskProcException(Exception): + ... + + +class TaskProcFatalException(TaskProcException): + ... + + +class TaskProcClientException(TaskProcException): + ... + + class AppRole(str, Enum): NORMAL = 'normal' INDEXING = 'indexing' @@ -169,6 +171,11 @@ class CommonSourceItem(BaseModel): provider: Annotated[str, AfterValidator(_validate_provider_id)] size: float + @computed_field + @property + def file_id(self) -> int: + return _get_file_id_from_source_ref(self.reference) + @field_validator('modified', mode='before') @classmethod def validate_modified(cls, v): @@ -205,11 +212,6 @@ def validate_type(self) -> Self: class ReceivedFileItem(CommonSourceItem): content: None - @computed_field - @property - def file_id(self) -> int: - return _get_file_id_from_source_ref(self.reference) - class SourceItem(CommonSourceItem): ''' diff --git a/context_chat_backend/utils.py b/context_chat_backend/utils.py index d5727140..fdfbf5da 100644 --- a/context_chat_backend/utils.py +++ b/context_chat_backend/utils.py @@ -14,7 +14,7 @@ from contextlib import suppress from functools import partial, wraps from multiprocessing.connection import Connection -from time import perf_counter_ns +from time import perf_counter_ns, time from typing import Any, TypeGuard, TypeVar from fastapi.responses import JSONResponse as FastAPIJSONResponse @@ -217,10 +217,13 @@ def exec_in_proc(group=None, target=None, name=None, args=(), kwargs=None, *, da stdobj = std_pconn.recv() # no need to update got_std here if stdobj.get('stdout') or stdobj.get('stderr'): - _logger.info('std info for %s', target_name, extra={ - 'stdout': stdobj.get('stdout', ''), - 'stderr': stdobj.get('stderr', ''), - }) + # log as WARNING if stderr is present + _logger.log((logging.INFO, logging.WARNING)[bool(stdobj.get('stderr'))], + 'std info for %s\nstdout: %s\nstderr: %s\n--------------------------', + target_name, + stdobj.get('stdout', '(none)'), + stdobj.get('stderr', '(none)'), + ) if not got_result: with suppress(EOFError, OSError, BrokenPipeError): @@ -320,3 +323,20 @@ def get_app_role() -> AppRole: def is_k8s_env(): role = get_app_role() return role != AppRole.NORMAL + + +# does not support caching of kwargs for recall +def timed_cache_async(ttl: int): + def decorator(fn: Callable): + cached_store: dict[tuple, tuple[float, Any]] = {} + @wraps(fn) + async def wrapper(*args, **kwargs): + if args in cached_store: + cached_time, cached_value = cached_store[args] + if (time() - cached_time) < ttl: + return cached_value + new_val = await fn(*args, **kwargs) + cached_store[args] = (time(), new_val) + return new_val + return wrapper + return decorator diff --git a/tests/ex_app/appinfo/info.xml b/tests/ex_app/appinfo/info.xml new file mode 100644 index 00000000..604ab2fb --- /dev/null +++ b/tests/ex_app/appinfo/info.xml @@ -0,0 +1,25 @@ + + + + ccb_test_providers + CCB Test Providers + Mock STT and OCR providers for CCB integration tests + 1.0.0 + AGPL-3.0-or-later + Nextcloud + CcbTestProviders + ai + + + + + + ghcr.io + nextcloud/ccb_test_providers + 1.0.0 + + + diff --git a/tests/ex_app/example.env b/tests/ex_app/example.env new file mode 100644 index 00000000..78bc7a38 --- /dev/null +++ b/tests/ex_app/example.env @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +# AppAPI headers +AA_VERSION=3.0.0 +APP_SECRET=12345 +APP_ID=ccb_test_providers +APP_DISPLAY_NAME=CCB Test Providers +APP_VERSION=1.0.0 +APP_HOST=0.0.0.0 +APP_PORT=10020 +APP_PERSISTENT_STORAGE=persistent_storage +#NEXTCLOUD_URL=http://localhost:8080 +NEXTCLOUD_URL=http://Nextcloud.nc.ncweb diff --git a/tests/ex_app/lib/main.py b/tests/ex_app/lib/main.py new file mode 100644 index 00000000..a746e175 --- /dev/null +++ b/tests/ex_app/lib/main.py @@ -0,0 +1,308 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +""" +Mock STT + OCR ex_app for CCB integration tests. + +Fake test files have the structure: + {magic_bytes}{identifier}\x00{null padding to 50 KB} + +The identifier (filename without extension) is used to look up the pre-stored +transcript/OCR text in TRANSCRIPTS. The magic bytes are skipped based on the +detected MIME type so the provider never tries to parse them as text. +""" +import logging +import os +import threading +import traceback +from contextlib import asynccontextmanager +from threading import Event +from time import sleep + +import magic as libmagic +from dotenv import load_dotenv +from fastapi import FastAPI + +load_dotenv() + +# ruff: noqa: E402 +from nc_py_api import AsyncNextcloudApp, NextcloudApp +from nc_py_api.ex_app import AppAPIAuthMiddleware, run_app, set_handlers +from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', +) +LOGGER = logging.getLogger(os.environ.get('APP_ID', 'ccb_test_providers')) +LOGGER.setLevel(logging.DEBUG) + +# --------------------------------------------------------------------------- +# Number of magic bytes written before the identifier in each fake file. +# Determined by the generator script (tests/gen_test_files.py). +# --------------------------------------------------------------------------- +MAGIC_LENGTHS: dict[str, int] = { + 'audio/mpeg': 4, # MP3 0xfffb9000 + 'audio/flac': 4, # fLaC + 'audio/x-m4a': 16, # ftyp M4A_ box + 'audio/x-hx-aac-adts': 4, # ADTS AAC + 'audio/x-wav': 16, # RIFF...WAVEfmt_ + 'image/png': 33, # PNG sig + IHDR chunk + 'image/jpeg': 11, # JFIF JPEG + 'image/gif': 6, # GIF89a + 'image/bmp': 54, # BMP file + DIB header + 'image/tiff': 4, # TIFF II/MM + 'image/webp': 12, # RIFF...WEBP + 'image/heic': 16, # ftyp heic box +} + +# --------------------------------------------------------------------------- +# Pre-stored transcripts / OCR results keyed by identifier (= filename stem). +# --------------------------------------------------------------------------- +TRANSCRIPTS: dict[str, str] = { + # Audio (STT) + 'amazon_rainforest': ( + 'The Amazon rainforest spans nine countries in South America ' + 'and produces roughly twenty percent of the world oxygen supply.' + ), + 'black_holes': ( + 'A black hole is a region of spacetime where gravity is so strong ' + 'that nothing, not even light, can escape from it.' + ), + 'continental_drift': ( + 'Continental drift is the gradual movement of the Earth landmasses ' + 'relative to each other over geological time scales.' + ), + 'fibonacci_sequence': ( + 'The Fibonacci sequence is a series of numbers where each number ' + 'is the sum of the two preceding ones, starting from zero and one.' + ), + 'jungle_book': ( + 'The Jungle Book follows Mowgli, a boy raised by wolves in the Indian jungle, ' + 'who must confront the tiger Shere Khan with the help of Baloo and Bagheera.' + ), + 'light_spectrum': ( + 'The visible light spectrum consists of wavelengths ranging from ' + 'approximately three hundred eighty to seven hundred nanometers.' + ), + 'monsoon_winds': ( + 'Monsoon winds are seasonal winds that bring heavy rainfall ' + 'to South Asia between June and September each year.' + ), + 'photosynthesis': ( + 'Photosynthesis is the process by which plants use sunlight, water, ' + 'and carbon dioxide to produce oxygen and energy in the form of sugar.' + ), + 'prime_numbers': ( + 'A prime number is a natural number greater than one ' + 'that has no positive divisors other than one and itself.' + ), + 'quantum_computing': ( + 'Quantum computing uses quantum mechanical phenomena such as ' + 'superposition and entanglement to perform calculations.' + ), + 'solar_system': ( + 'The solar system consists of the Sun and all objects that orbit it, ' + 'including eight planets, dwarf planets, moons, and asteroids.' + ), + 'tcp_ip_overview': ( + 'The TCP IP protocol suite is the foundational communication protocol ' + 'of the Internet, enabling data exchange across diverse networks.' + ), + + # Image (OCR) + 'book_page': ( + 'Chapter One: The Origins of Writing. ' + 'Writing emerged independently in several ancient civilizations around five thousand years ago.' + ), + 'diagram_labels': ( + 'Figure 1: System Architecture. ' + 'Components: Input Layer, Processing Unit, Output Buffer, Storage Module.' + ), + 'invoice_sample': ( + 'Invoice No. 10042. Date: 01/05/2026. ' + 'Item: Software License. Quantity: 1. Unit Price: 299.00. Total: 299.00.' + ), + 'map_legend': ( + 'Legend: Blue = Water bodies. Green = Forest. ' + 'Yellow = Agricultural land. Red = Urban area. Scale 1:50000.' + ), + 'newspaper_headline': ( + 'Scientists Discover New Method for Carbon Capture Using Algae Bioreactors. ' + 'Research published in Nature, April 2026.' + ), + 'product_label': ( + 'Organic Green Tea. Net weight 100g. Ingredients: Green tea leaves. ' + 'Store in a cool dry place. Best before: 2027.' + ), + 'shop_receipt': ( + 'RECEIPT. Store: Central Market. ' + 'Items: Rice 2kg x1.50, Bread x0.90, Milk 1L x1.20. Total: 3.60. Thank you.' + ), + 'street_sign': ( + 'MARKET STREET. Speed limit 30. ' + 'No parking 8am to 6pm Monday to Saturday.' + ), + 'test': "This is a test for Nextcloud's OCR app.", + 'train_schedule': ( + 'Departures: Platform 3. ' + '08:14 Express to Amsterdam. 08:32 Regional to Utrecht. 08:55 Intercity to Rotterdam.' + ), +} + +STT_PROVIDER_ID = 'ccb_test_providers:stt' +STT_TASK_TYPE_ID = 'core:audio2text' +OCR_PROVIDER_ID = 'ccb_test_providers:ocr' +OCR_TASK_TYPE_ID = 'core:image2text:ocr' + +app_enabled = Event() +TRIGGER = Event() +WAIT_INTERVAL = 5 +WAIT_INTERVAL_WITH_TRIGGER = 5 * 60 + + +def _fetch_file_header(nc: NextcloudApp, task_id: int, file_id: int) -> bytes: + """Fetch only the first 256 bytes of a task file (enough to extract the identifier).""" + nc._session.init_adapter() + resp = nc._session.adapter.request( + 'GET', + f'/ocs/v2.php/taskprocessing/tasks_provider/{task_id}/file/{file_id}', + headers={'Range': 'bytes=0-255'}, + ) + if resp.status_code not in (200, 206): + raise Exception(f'Failed to fetch file {file_id} for task {task_id}: HTTP {resp.status_code}') + return resp.content + + +def _transcript_for_bytes(raw: bytes) -> str: + mime = libmagic.from_buffer(raw[:256], mime=True) + offset = MAGIC_LENGTHS.get(mime) + if offset is None: + raise ValueError(f'Unsupported MIME type for fake file: {mime!r}') + end = raw.index(b'\x00', offset) + identifier = raw[offset:end].decode('ascii') + text = TRANSCRIPTS.get(identifier) + if text is None: + raise ValueError(f'No transcript registered for identifier: {identifier!r}') + return text + + +# --------------------------------------------------------------------------- +# Background worker - handles both task types in one loop +# --------------------------------------------------------------------------- + +def background_thread_task(): + nc = NextcloudApp() + + while True: + if not app_enabled.is_set(): + LOGGER.debug('App is not enabled, sleeping for 5 secs') + sleep(5) + continue + + # STT + try: + item = nc.providers.task_processing.next_task( + [STT_PROVIDER_ID], [STT_TASK_TYPE_ID] + ) + if item and 'task' in item: + task = item['task'] + try: + LOGGER.info(f'STT task {task["id"]}') + raw = _fetch_file_header(nc, task['id'], task['input']['input']) + transcript = _transcript_for_bytes(raw) + nc.providers.task_processing.report_result(task['id'], {'output': transcript}) + LOGGER.info(f'STT task {task["id"]} done') + except Exception as e: + LOGGER.error(traceback.format_exc()) + try: + nc.providers.task_processing.report_result(task['id'], None, str(e)) + except Exception: # noqa: S110 + pass + continue # skip wait, poll again immediately + except Exception as e: + LOGGER.error(f'Error polling STT: {e}') + + # OCR + try: + item = nc.providers.task_processing.next_task( + [OCR_PROVIDER_ID], [OCR_TASK_TYPE_ID] + ) + if item and 'task' in item: + task = item['task'] + try: + LOGGER.info(f'OCR task {task["id"]}') + file_ids: list[int] = task['input']['input'] + outputs = [ + _transcript_for_bytes(_fetch_file_header(nc, task['id'], fid)) + for fid in file_ids + ] + nc.providers.task_processing.report_result(task['id'], {'output': outputs}) + LOGGER.info(f'OCR task {task["id"]} done') + except Exception as e: + LOGGER.error(traceback.format_exc()) + try: + nc.providers.task_processing.report_result(task['id'], None, str(e)) + except Exception: # noqa: S110 + pass + continue + except Exception as e: + LOGGER.error(f'Error polling OCR: {e}') + + _wait_for_task() + + +def _wait_for_task(interval: float | None = None): + global WAIT_INTERVAL, WAIT_INTERVAL_WITH_TRIGGER + if interval is None: + interval = WAIT_INTERVAL + if TRIGGER.wait(timeout=interval): + WAIT_INTERVAL = WAIT_INTERVAL_WITH_TRIGGER + TRIGGER.clear() + + +# --------------------------------------------------------------------------- +# FastAPI app + lifecycle +# --------------------------------------------------------------------------- + +@asynccontextmanager +async def lifespan(_app: FastAPI): + set_handlers(_app, enabled_handler, trigger_handler=trigger_handler) + nc = NextcloudApp() + if nc.enabled_state: + app_enabled.set() + threading.Thread(target=background_thread_task, daemon=True).start() + yield + + +APP = FastAPI(lifespan=lifespan) +APP.add_middleware(AppAPIAuthMiddleware) + + +async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str: + if enabled: + await nc.providers.task_processing.register(TaskProcessingProvider( + id=STT_PROVIDER_ID, + name='CCB Test STT Provider', + task_type=STT_TASK_TYPE_ID, + expected_runtime=5, + )) + await nc.providers.task_processing.register(TaskProcessingProvider( + id=OCR_PROVIDER_ID, + name='CCB Test OCR Provider', + task_type=OCR_TASK_TYPE_ID, + expected_runtime=5, + )) + app_enabled.set() + else: + await nc.providers.task_processing.unregister(STT_PROVIDER_ID, True) + await nc.providers.task_processing.unregister(OCR_PROVIDER_ID, True) + app_enabled.clear() + return '' + + +def trigger_handler(provider_id: str): + TRIGGER.set() + + +if __name__ == '__main__': + run_app('main:APP', log_level='info') diff --git a/tests/res/.gitattributes b/tests/res/.gitattributes new file mode 100644 index 00000000..ab30abe3 --- /dev/null +++ b/tests/res/.gitattributes @@ -0,0 +1,16 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +# Binary test files - prevent line-ending conversion and diff noise +*.aac binary +*.bmp binary +*.flac binary +*.gif binary +*.heic binary +*.jpg binary +*.m4a binary +*.mp3 binary +*.png binary +*.tiff binary +*.wav binary +*.webp binary diff --git a/tests/res/audio/amazon_rainforest.aac b/tests/res/audio/amazon_rainforest.aac new file mode 100644 index 00000000..966a49f3 Binary files /dev/null and b/tests/res/audio/amazon_rainforest.aac differ diff --git a/tests/res/audio/black_holes.aac b/tests/res/audio/black_holes.aac new file mode 100644 index 00000000..4209571f Binary files /dev/null and b/tests/res/audio/black_holes.aac differ diff --git a/tests/res/audio/continental_drift.m4a b/tests/res/audio/continental_drift.m4a new file mode 100644 index 00000000..4915e60b Binary files /dev/null and b/tests/res/audio/continental_drift.m4a differ diff --git a/tests/res/audio/fibonacci_sequence.wav b/tests/res/audio/fibonacci_sequence.wav new file mode 100644 index 00000000..dc459298 Binary files /dev/null and b/tests/res/audio/fibonacci_sequence.wav differ diff --git a/tests/res/audio/jungle_book.m4a b/tests/res/audio/jungle_book.m4a new file mode 100644 index 00000000..be8c6efd Binary files /dev/null and b/tests/res/audio/jungle_book.m4a differ diff --git a/tests/res/audio/light_spectrum.wav b/tests/res/audio/light_spectrum.wav new file mode 100644 index 00000000..0f39813e Binary files /dev/null and b/tests/res/audio/light_spectrum.wav differ diff --git a/tests/res/audio/monsoon_winds.wav b/tests/res/audio/monsoon_winds.wav new file mode 100644 index 00000000..8be769bf Binary files /dev/null and b/tests/res/audio/monsoon_winds.wav differ diff --git a/tests/res/audio/photosynthesis.mp3 b/tests/res/audio/photosynthesis.mp3 new file mode 100644 index 00000000..e17816ce Binary files /dev/null and b/tests/res/audio/photosynthesis.mp3 differ diff --git a/tests/res/audio/prime_numbers.mp3 b/tests/res/audio/prime_numbers.mp3 new file mode 100644 index 00000000..678b2858 Binary files /dev/null and b/tests/res/audio/prime_numbers.mp3 differ diff --git a/tests/res/audio/quantum_computing.flac b/tests/res/audio/quantum_computing.flac new file mode 100644 index 00000000..a568541c Binary files /dev/null and b/tests/res/audio/quantum_computing.flac differ diff --git a/tests/res/audio/solar_system.flac b/tests/res/audio/solar_system.flac new file mode 100644 index 00000000..800186c8 Binary files /dev/null and b/tests/res/audio/solar_system.flac differ diff --git a/tests/res/audio/tcp_ip_overview.mp3 b/tests/res/audio/tcp_ip_overview.mp3 new file mode 100644 index 00000000..b9a67174 Binary files /dev/null and b/tests/res/audio/tcp_ip_overview.mp3 differ diff --git a/tests/res/image/book_page.tiff b/tests/res/image/book_page.tiff new file mode 100644 index 00000000..b2cb7d4c Binary files /dev/null and b/tests/res/image/book_page.tiff differ diff --git a/tests/res/image/diagram_labels.gif b/tests/res/image/diagram_labels.gif new file mode 100644 index 00000000..3d6c5656 Binary files /dev/null and b/tests/res/image/diagram_labels.gif differ diff --git a/tests/res/image/invoice_sample.png b/tests/res/image/invoice_sample.png new file mode 100644 index 00000000..108230ba Binary files /dev/null and b/tests/res/image/invoice_sample.png differ diff --git a/tests/res/image/map_legend.png b/tests/res/image/map_legend.png new file mode 100644 index 00000000..338ea3e2 Binary files /dev/null and b/tests/res/image/map_legend.png differ diff --git a/tests/res/image/newspaper_headline.webp b/tests/res/image/newspaper_headline.webp new file mode 100644 index 00000000..b2769904 Binary files /dev/null and b/tests/res/image/newspaper_headline.webp differ diff --git a/tests/res/image/product_label.heic b/tests/res/image/product_label.heic new file mode 100644 index 00000000..9118bcca Binary files /dev/null and b/tests/res/image/product_label.heic differ diff --git a/tests/res/image/shop_receipt.bmp b/tests/res/image/shop_receipt.bmp new file mode 100644 index 00000000..2e4dac09 Binary files /dev/null and b/tests/res/image/shop_receipt.bmp differ diff --git a/tests/res/image/street_sign.jpg b/tests/res/image/street_sign.jpg new file mode 100644 index 00000000..3f0291c6 Binary files /dev/null and b/tests/res/image/street_sign.jpg differ diff --git a/tests/res/image/test.png b/tests/res/image/test.png new file mode 100644 index 00000000..a013e6c3 Binary files /dev/null and b/tests/res/image/test.png differ diff --git a/tests/res/image/train_schedule.jpg b/tests/res/image/train_schedule.jpg new file mode 100644 index 00000000..43fcf6ce Binary files /dev/null and b/tests/res/image/train_schedule.jpg differ