open calls in paperless-ngx

gistfile1.txt

$ grep -Rn -C 3 --exclude-dir=tests --exclude-dir=migrations "open(" src/
src/paperless/views.py-82-            "img",
src/paperless/views.py-83-            "favicon.ico",
src/paperless/views.py-84-        )
src/paperless/views.py:85:        with open(favicon, "rb") as f:
src/paperless/views.py-86-            return HttpResponse(f, content_type="image/x-icon")
src/paperless/views.py-87-
src/paperless/views.py-88-
--
src/paperless/checks.py-32-                f"__paperless_write_test_{os.getpid()}__",
src/paperless/checks.py-33-            )
src/paperless/checks.py-34-            try:
src/paperless/checks.py:35:                with open(test_file, "w"):
src/paperless/checks.py-36-                    pass
src/paperless/checks.py-37-            except PermissionError:
src/paperless/checks.py-38-                dir_stat = os.stat(directory)
--
src/paperless_tesseract/parsers.py-45-        if mime_type == "application/pdf":
src/paperless_tesseract/parsers.py-46-            import pikepdf
src/paperless_tesseract/parsers.py-47-
src/paperless_tesseract/parsers.py:48:            with pikepdf.Pdf.open(document_path) as pdf:
src/paperless_tesseract/parsers.py-49-                page_count = len(pdf.pages)
src/paperless_tesseract/parsers.py-50-        return page_count
src/paperless_tesseract/parsers.py-51-
--
src/paperless_tesseract/parsers.py-56-
src/paperless_tesseract/parsers.py-57-            namespace_pattern = re.compile(r"\{(.*)\}(.*)")
src/paperless_tesseract/parsers.py-58-
src/paperless_tesseract/parsers.py:59:            pdf = pikepdf.open(document_path)
src/paperless_tesseract/parsers.py-60-            meta = pdf.open_metadata()
src/paperless_tesseract/parsers.py-61-            for key, value in meta.items():
src/paperless_tesseract/parsers.py-62-                if isinstance(value, list):
--
src/paperless_tesseract/parsers.py-106-        ]
src/paperless_tesseract/parsers.py-107-
src/paperless_tesseract/parsers.py-108-    def has_alpha(self, image) -> bool:
src/paperless_tesseract/parsers.py:109:        with Image.open(image) as im:
src/paperless_tesseract/parsers.py-110-            return im.mode in ("RGBA", "LA")
src/paperless_tesseract/parsers.py-111-
src/paperless_tesseract/parsers.py-112-    def remove_alpha(self, image_path: str) -> Path:
--
src/paperless_tesseract/parsers.py-125-
src/paperless_tesseract/parsers.py-126-    def get_dpi(self, image) -> int | None:
src/paperless_tesseract/parsers.py-127-        try:
src/paperless_tesseract/parsers.py:128:            with Image.open(image) as im:
src/paperless_tesseract/parsers.py-129-                x, y = im.info["dpi"]
src/paperless_tesseract/parsers.py-130-                return round(x)
src/paperless_tesseract/parsers.py-131-        except Exception as e:
--
src/paperless_tesseract/parsers.py-134-
src/paperless_tesseract/parsers.py-135-    def calculate_a4_dpi(self, image) -> int | None:
src/paperless_tesseract/parsers.py-136-        try:
src/paperless_tesseract/parsers.py:137:            with Image.open(image) as im:
src/paperless_tesseract/parsers.py-138-                width, height = im.size
src/paperless_tesseract/parsers.py-139-                # divide image width by A4 width (210mm) in inches.
src/paperless_tesseract/parsers.py-140-                dpi = int(width / (21 / 2.54))
--
src/paperless_mail/parsers.py-188-        Parses the given .eml file into a MailMessage object
src/paperless_mail/parsers.py-189-        """
src/paperless_mail/parsers.py-190-        try:
src/paperless_mail/parsers.py:191:            with filepath.open("rb") as eml:
src/paperless_mail/parsers.py-192-                parsed = MailMessage.from_bytes(eml.read())
src/paperless_mail/parsers.py-193-                if parsed.from_values is None:
src/paperless_mail/parsers.py-194-                    raise ParseError(
--
src/paperless_mail/mail.py-900-            dir=settings.SCRATCH_DIR,
src/paperless_mail/mail.py-901-            suffix=".eml",
src/paperless_mail/mail.py-902-        )
src/paperless_mail/mail.py:903:        with open(temp_filename, "wb") as f:
src/paperless_mail/mail.py-904-            # Move "From"-header to beginning of file
src/paperless_mail/mail.py-905-            # TODO: This ugly workaround is needed because the parser is
src/paperless_mail/mail.py-906-            #   chosen only by the mime_type detected via magic
--
src/documents/barcodes.py-231-
src/documents/barcodes.py-232-        try:
src/documents/barcodes.py-233-            # Read number of pages from pdf
src/documents/barcodes.py:234:            with Pdf.open(self.pdf_file) as pdf:
src/documents/barcodes.py-235-                num_of_pages = len(pdf.pages)
src/documents/barcodes.py-236-            logger.debug(f"PDF has {num_of_pages} pages")
src/documents/barcodes.py-237-
--
src/documents/barcodes.py-414-
src/documents/barcodes.py-415-        document_paths = []
src/documents/barcodes.py-416-        fname = self.input_doc.original_file.stem
src/documents/barcodes.py:417:        with Pdf.open(self.pdf_file) as input_pdf:
src/documents/barcodes.py-418-            # Start with an empty document
src/documents/barcodes.py-419-            current_document: list[Page] = []
src/documents/barcodes.py-420-            # A list of documents, ie a list of lists of pages
--
src/documents/barcodes.py-451-
src/documents/barcodes.py-452-                logger.debug(f"pdf no:{doc_idx} has {len(dst.pages)} pages")
src/documents/barcodes.py-453-                savepath = Path(self.temp_dir.name) / output_filename
src/documents/barcodes.py:454:                with open(savepath, "wb") as out:
src/documents/barcodes.py-455-                    dst.save(out)
src/documents/barcodes.py-456-
src/documents/barcodes.py-457-                copy_basic_file_stats(self.input_doc.original_file, savepath)
--
src/documents/classifier.py-95-    def load(self) -> None:
src/documents/classifier.py-96-        # Catch warnings for processing
src/documents/classifier.py-97-        with warnings.catch_warnings(record=True) as w:
src/documents/classifier.py:98:            with open(settings.MODEL_FILE, "rb") as f:
src/documents/classifier.py-99-                schema_version = pickle.load(f)
src/documents/classifier.py-100-
src/documents/classifier.py-101-                if schema_version != self.FORMAT_VERSION:
--
src/documents/classifier.py-136-        target_file: Path = settings.MODEL_FILE
src/documents/classifier.py-137-        target_file_temp = target_file.with_suffix(".pickle.part")
src/documents/classifier.py-138-
src/documents/classifier.py:139:        with open(target_file_temp, "wb") as f:
src/documents/classifier.py-140-            pickle.dump(self.FORMAT_VERSION, f)
src/documents/classifier.py-141-
src/documents/classifier.py-142-            pickle.dump(self.last_doc_change_time, f)
--
src/documents/consumer.py-164-        """
src/documents/consumer.py-165-        Using the MD5 of the file, check this exact file doesn't already exist
src/documents/consumer.py-166-        """
src/documents/consumer.py:167:        with open(self.input_doc.original_file, "rb") as f:
src/documents/consumer.py-168-            checksum = hashlib.md5(f.read()).hexdigest()
src/documents/consumer.py-169-        existing_doc = Document.global_objects.filter(
src/documents/consumer.py-170-            Q(checksum=checksum) | Q(archive_checksum=checksum),
--
src/documents/consumer.py-590-                            document.archive_path,
src/documents/consumer.py-591-                        )
src/documents/consumer.py-592-
src/documents/consumer.py:593:                        with open(archive_path, "rb") as f:
src/documents/consumer.py-594-                            document.archive_checksum = hashlib.md5(
src/documents/consumer.py-595-                                f.read(),
src/documents/consumer.py-596-                            ).hexdigest()
--
src/documents/consumer.py-789-                )  # adds to document
src/documents/consumer.py-790-
src/documents/consumer.py-791-    def _write(self, storage_type, source, target):
src/documents/consumer.py:792:        with open(source, "rb") as read_file, open(target, "wb") as write_file:
src/documents/consumer.py-793-            write_file.write(read_file.read())
src/documents/consumer.py-794-
src/documents/consumer.py-795-        # Attempt to copy file's original stats, but it's ok if we can't
--
src/documents/double_sided.py-87-        if valid_staging_exists:
src/documents/double_sided.py-88-            try:
src/documents/double_sided.py-89-                # Collate pages from second PDF in reverse order
src/documents/double_sided.py:90:                with Pdf.open(staging) as pdf1, Pdf.open(pdf_file) as pdf2:
src/documents/double_sided.py-91-                    pdf2.pages.reverse()
src/documents/double_sided.py-92-                    try:
src/documents/double_sided.py-93-                        for i, page in enumerate(pdf2.pages):
--
src/documents/bulk_edit.py-221-            )
src/documents/bulk_edit.py-222-            continue
src/documents/bulk_edit.py-223-        try:
src/documents/bulk_edit.py:224:            with pikepdf.open(doc.source_path, allow_overwriting_input=True) as pdf:
src/documents/bulk_edit.py-225-                for page in pdf.pages:
src/documents/bulk_edit.py-226-                    page.rotate(degrees, relative=True)
src/documents/bulk_edit.py-227-                pdf.save()
--
src/documents/bulk_edit.py-265-    for doc_id in doc_ids:
src/documents/bulk_edit.py-266-        doc = qs.get(id=doc_id)
src/documents/bulk_edit.py-267-        try:
src/documents/bulk_edit.py:268:            with pikepdf.open(str(doc.source_path)) as pdf:
src/documents/bulk_edit.py-269-                version = max(version, pdf.pdf_version)
src/documents/bulk_edit.py-270-                merged_pdf.pages.extend(pdf.pages)
src/documents/bulk_edit.py-271-            affected_docs.append(doc.id)
--
src/documents/bulk_edit.py-332-    consume_tasks = []
src/documents/bulk_edit.py-333-
src/documents/bulk_edit.py-334-    try:
src/documents/bulk_edit.py:335:        with pikepdf.open(doc.source_path) as pdf:
src/documents/bulk_edit.py-336-            for idx, split_doc in enumerate(pages):
src/documents/bulk_edit.py-337-                dst = pikepdf.new()
src/documents/bulk_edit.py-338-                for page in split_doc:
--
src/documents/bulk_edit.py-385-    import pikepdf
src/documents/bulk_edit.py-386-
src/documents/bulk_edit.py-387-    try:
src/documents/bulk_edit.py:388:        with pikepdf.open(doc.source_path, allow_overwriting_input=True) as pdf:
src/documents/bulk_edit.py-389-            offset = 1  # pages are 1-indexed
src/documents/bulk_edit.py-390-            for page_num in pages:
src/documents/bulk_edit.py-391-                pdf.pages.remove(pdf.pages[page_num - offset])
--
src/documents/views.py-928-        if not os.path.isfile(filename):
src/documents/views.py-929-            raise Http404
src/documents/views.py-930-
src/documents/views.py:931:        with open(filename) as f:
src/documents/views.py-932-            lines = [line.rstrip() for line in f.readlines()]
src/documents/views.py-933-
src/documents/views.py-934-        return Response(lines)
--
src/documents/views.py-1517-            for document in Document.objects.filter(pk__in=ids):
src/documents/views.py-1518-                strategy.add_document(document)
src/documents/views.py-1519-
src/documents/views.py:1520:        with open(temp.name, "rb") as f:
src/documents/views.py-1521-            response = HttpResponse(f, content_type="application/zip")
src/documents/views.py-1522-            response["Content-Disposition"] = '{}; filename="{}"'.format(
src/documents/views.py-1523-                "attachment",
--
src/documents/views.py-1674-            # Ensure a JSON response
src/documents/views.py-1675-            req.add_header("Accept", "application/json")
src/documents/views.py-1676-
src/documents/views.py:1677:            with urllib.request.urlopen(req) as response:
src/documents/views.py-1678-                remote = response.read().decode("utf8")
src/documents/views.py-1679-            try:
src/documents/views.py-1680-                remote_json = json.loads(remote)
--
src/documents/management/commands/decrypt_documents.py-79-
src/documents/management/commands/decrypt_documents.py-80-            document.filename = os.path.splitext(document.filename)[0]
src/documents/management/commands/decrypt_documents.py-81-
src/documents/management/commands/decrypt_documents.py:82:            with open(document.source_path, "wb") as f:
src/documents/management/commands/decrypt_documents.py-83-                f.write(raw_document)
src/documents/management/commands/decrypt_documents.py-84-
src/documents/management/commands/decrypt_documents.py:85:            with open(document.thumbnail_path, "wb") as f:
src/documents/management/commands/decrypt_documents.py-86-                f.write(raw_thumb)
src/documents/management/commands/decrypt_documents.py-87-
src/documents/management/commands/decrypt_documents.py-88-            Document.objects.filter(id=document.id).update(
--
src/documents/management/commands/document_importer.py-143-        """
src/documents/management/commands/document_importer.py-144-        main_manifest_path = self.source / "manifest.json"
src/documents/management/commands/document_importer.py-145-
src/documents/management/commands/document_importer.py:146:        with main_manifest_path.open() as infile:
src/documents/management/commands/document_importer.py-147-            self.manifest = json.load(infile)
src/documents/management/commands/document_importer.py-148-        self.manifest_paths.append(main_manifest_path)
src/documents/management/commands/document_importer.py-149-
src/documents/management/commands/document_importer.py-150-        for file in Path(self.source).glob("**/*-manifest.json"):
src/documents/management/commands/document_importer.py:151:            with file.open() as infile:
src/documents/management/commands/document_importer.py-152-                self.manifest += json.load(infile)
src/documents/management/commands/document_importer.py-153-            self.manifest_paths.append(file)
src/documents/management/commands/document_importer.py-154-
--
src/documents/management/commands/document_importer.py-167-            return
src/documents/management/commands/document_importer.py-168-
src/documents/management/commands/document_importer.py-169-        if metadata_path.exists():
src/documents/management/commands/document_importer.py:170:            with metadata_path.open() as infile:
src/documents/management/commands/document_importer.py-171-                data = json.load(infile)
src/documents/management/commands/document_importer.py-172-                self.version = data["version"]
src/documents/management/commands/document_importer.py-173-                if not self.passphrase and EXPORTER_CRYPTO_SETTINGS_NAME in data:
--
src/documents/management/commands/document_importer.py-177-                elif EXPORTER_CRYPTO_SETTINGS_NAME in data:
src/documents/management/commands/document_importer.py-178-                    self.load_crypt_params(data)
src/documents/management/commands/document_importer.py-179-        elif version_path.exists():
src/documents/management/commands/document_importer.py:180:            with version_path.open() as infile:
src/documents/management/commands/document_importer.py-181-                self.version = json.load(infile)["version"]
src/documents/management/commands/document_importer.py-182-
src/documents/management/commands/document_importer.py-183-        if self.version and self.version != version.__full_version_str__:
--
src/documents/management/commands/document_importer.py-311-                    "appear to be in the source directory.",
src/documents/management/commands/document_importer.py-312-                )
src/documents/management/commands/document_importer.py-313-            try:
src/documents/management/commands/document_importer.py:314:                with doc_path.open(mode="rb"):
src/documents/management/commands/document_importer.py-315-                    pass
src/documents/management/commands/document_importer.py-316-            except Exception as e:
src/documents/management/commands/document_importer.py-317-                raise CommandError(
--
src/documents/management/commands/document_importer.py-327-                        f"does not appear to be in the source directory.",
src/documents/management/commands/document_importer.py-328-                    )
src/documents/management/commands/document_importer.py-329-                try:
src/documents/management/commands/document_importer.py:330:                    with doc_archive_path.open(mode="rb"):
src/documents/management/commands/document_importer.py-331-                        pass
src/documents/management/commands/document_importer.py-332-                except Exception as e:
src/documents/management/commands/document_importer.py-333-                    raise CommandError(
--
src/documents/management/commands/document_consumer.py-107-
src/documents/management/commands/document_consumer.py-108-    while (read_try_count < os_error_retry_count) and not file_open_ok:
src/documents/management/commands/document_consumer.py-109-        try:
src/documents/management/commands/document_consumer.py:110:            with open(filepath, "rb"):
src/documents/management/commands/document_consumer.py-111-                file_open_ok = True
src/documents/management/commands/document_consumer.py-112-        except OSError as e:
src/documents/management/commands/document_consumer.py-113-            read_try_count += 1
--
src/documents/plugins/helpers.py-27-        self.task_id = task_id
src/documents/plugins/helpers.py-28-
src/documents/plugins/helpers.py-29-    def __enter__(self):
src/documents/plugins/helpers.py:30:        self.open()
src/documents/plugins/helpers.py-31-        return self
src/documents/plugins/helpers.py-32-
src/documents/plugins/helpers.py-33-    def __exit__(self, exc_type, exc_val, exc_tb):
src/documents/plugins/helpers.py-34-        self.close()
src/documents/plugins/helpers.py-35-
src/documents/plugins/helpers.py:36:    def open(self) -> None:
src/documents/plugins/helpers.py-37-        """
src/documents/plugins/helpers.py-38-        If not already opened, gets the default channel layer
src/documents/plugins/helpers.py-39-        opened and ready to send messages
--
src/documents/plugins/helpers.py-58-        extra_args: dict[str, str | int | None] | None = None,
src/documents/plugins/helpers.py-59-    ) -> None:
src/documents/plugins/helpers.py-60-        # Ensure the layer is open
src/documents/plugins/helpers.py:61:        self.open()
src/documents/plugins/helpers.py-62-
src/documents/plugins/helpers.py-63-        # Just for IDEs
src/documents/plugins/helpers.py-64-        if TYPE_CHECKING:
--
src/documents/converters.py-21-    # override pixel setting if needed
src/documents/converters.py-22-    maybe_override_pixel_limit()
src/documents/converters.py-23-
src/documents/converters.py:24:    with Image.open(tiff_path) as im:
src/documents/converters.py-25-        has_alpha_layer = im.mode in ("RGBA", "LA")
src/documents/converters.py-26-    if has_alpha_layer:
src/documents/converters.py-27-        # Note the save into the temp folder, so as not to trigger a new
--
src/documents/converters.py-42-
src/documents/converters.py-43-    pdf_path = (target_directory / tiff_path.name).with_suffix(".pdf")
src/documents/converters.py-44-
src/documents/converters.py:45:    with scratch_image.open("rb") as img_file, pdf_path.open("wb") as pdf_file:
src/documents/converters.py-46-        pdf_file.write(img2pdf.convert(img_file))
src/documents/converters.py-47-
src/documents/converters.py-48-    # Copy what file stat is possible
--
src/documents/tasks.py-236-
src/documents/tasks.py-237-        if parser.get_archive_path():
src/documents/tasks.py-238-            with transaction.atomic():
src/documents/tasks.py:239:                with open(parser.get_archive_path(), "rb") as f:
src/documents/tasks.py-240-                    checksum = hashlib.md5(f.read()).hexdigest()
src/documents/tasks.py-241-                # I'm going to save first so that in case the file move
src/documents/tasks.py-242-                # fails, the database is rolled back.
--
src/documents/models.py-320-
src/documents/models.py-321-    @property
src/documents/models.py-322-    def source_file(self):
src/documents/models.py:323:        return open(self.source_path, "rb")
src/documents/models.py-324-
src/documents/models.py-325-    @property
src/documents/models.py-326-    def has_archive_version(self) -> bool:
--
src/documents/models.py-335-
src/documents/models.py-336-    @property
src/documents/models.py-337-    def archive_file(self):
src/documents/models.py:338:        return open(self.archive_path, "rb")
src/documents/models.py-339-
src/documents/models.py-340-    def get_public_filename(self, archive=False, counter=0, suffix=None) -> str:
src/documents/models.py-341-        """
--
src/documents/models.py-372-
src/documents/models.py-373-    @property
src/documents/models.py-374-    def thumbnail_file(self):
src/documents/models.py:375:        return open(self.thumbnail_path, "rb")
src/documents/models.py-376-
src/documents/models.py-377-    @property
src/documents/models.py-378-    def created_date(self):