annif.corpus.document.DocumentFileCSV._parse_row() - Code Metrics - Inspection of "WIP: Add JSONL short text corpus format" - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#876)

by Osma

created 2025-08-18 08:53 UTC

annif.corpus.document.DocumentFileCSV._parse_row() A

↳ Parent: annif.corpus.document

Complexity

Conditions

Size

Total Lines	12
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	10
nop	2
dl	0
loc	12
rs	9.9
c	0
b	0
f	0

"""Classes for supporting document corpora"""

from __future__ import annotations

import csv
import glob
import gzip
import os.path
import re
from itertools import islice
from typing import TYPE_CHECKING

import annif.util
from annif.exception import OperationFailedException

from .json import json_file_to_document, json_to_document
from .types import Document, DocumentCorpus, SubjectSet

if TYPE_CHECKING:
    from collections.abc import Iterator

    from annif.vocab import SubjectIndex

logger = annif.logger


class DocumentDirectory(DocumentCorpus):
    """A directory of files as a full text document corpus"""

    def __init__(
        self,
        path: str,
        subject_index: SubjectIndex | None = None,
        language: str | None = None,
        require_subjects: bool = False,
    ) -> None:
        self.path = path
        self.subject_index = subject_index
        self.language = language
        self.require_subjects = require_subjects

    def __iter__(self) -> Iterator[str]:
        """Iterate through the directory, yielding file paths with corpus documents."""

        # txt files
        for filename in sorted(glob.glob(os.path.join(self.path, "*.txt"))):
            yield filename

        # json files
        for filename in sorted(glob.glob(os.path.join(self.path, "*.json"))):
            yield filename

    @staticmethod
    def _get_subject_filename(filename: str) -> str | None:
        tsvfilename = re.sub(r"\.txt$", ".tsv", filename)
        if os.path.exists(tsvfilename):
            return tsvfilename

        keyfilename = re.sub(r"\.txt$", ".key", filename)
        if os.path.exists(keyfilename):
            return keyfilename

        return None

    def _read_txt_file(self, filename: str) -> Document | None:
        with open(filename, errors="replace", encoding="utf-8-sig") as docfile:
            text = docfile.read()
        if not self.require_subjects:
            return Document(text=text, subject_set=None, file_path=filename)

        subjfilename = self._get_subject_filename(filename)
        if subjfilename is None:
            # subjects required but not found, skipping this docfile
            return None

        with open(subjfilename, encoding="utf-8-sig") as subjfile:
            subjects = SubjectSet.from_string(
                subjfile.read(), self.subject_index, self.language
            )
        return Document(text=text, subject_set=subjects, file_path=filename)

    @property
    def documents(self) -> Iterator[Document]:
        for docfilename in self:
            if docfilename.endswith(".txt"):
                doc = self._read_txt_file(docfilename)
            else:
                doc = json_file_to_document(
                    docfilename,
                    self.subject_index,
                    self.language,
                    self.require_subjects,
                )

            if doc is not None:
                yield doc


class DocumentFileTSV(DocumentCorpus):
    """A TSV file as a corpus of documents with subjects"""

    def __init__(self, path: str, subject_index: SubjectIndex) -> None:
        self.path = path
        self.subject_index = subject_index

    @property
    def documents(self) -> Iterator[Document]:
        if self.path.endswith(".gz"):
            opener = gzip.open
        else:
            opener = open
        with opener(self.path, mode="rt", encoding="utf-8-sig") as tsvfile:
            for line in tsvfile:
                yield from self._parse_tsv_line(line)

    def _parse_tsv_line(self, line: str) -> Iterator[Document]:
        if "\t" in line:
            text, uris = line.split("\t", maxsplit=1)
            subject_ids = {
                self.subject_index.by_uri(annif.util.cleanup_uri(uri))
                for uri in uris.split()
            }
            yield Document(text=text, subject_set=SubjectSet(subject_ids))
        else:
            logger.warning('Skipping invalid line (missing tab): "%s"', line.rstrip())


class DocumentFileCSV(DocumentCorpus):
    """A CSV file as a corpus of documents with subjects"""

    def __init__(self, path: str, subject_index: SubjectIndex) -> None:
        self.path = path
        self.subject_index = subject_index

    @property
    def documents(self) -> Iterator[Document]:
        if self.path.endswith(".gz"):
            opener = gzip.open
        else:
            opener = open
        with opener(self.path, mode="rt", encoding="utf-8-sig") as csvfile:
            reader = csv.DictReader(csvfile)
            if not self._check_fields(reader):
                raise OperationFailedException(
                    f"Cannot parse CSV file {self.path}. "
                    + "The file must have a header row that defines at least "
                    + "the columns 'text' and 'subject_uris'."
                )
            for row in reader:
                yield from self._parse_row(row)

    def _parse_row(self, row: dict[str, str]) -> Iterator[Document]:
        subject_ids = {
            self.subject_index.by_uri(annif.util.cleanup_uri(uri))
            for uri in (row["subject_uris"] or "").strip().split()
        }
        metadata = {
            key: val for key, val in row.items() if key not in ("text", "subject_uris")
        }
        yield Document(
            text=(row["text"] or ""),
            subject_set=SubjectSet(subject_ids),
            metadata=metadata,
        )

    def _check_fields(self, reader: csv.DictReader) -> bool:
        fns = reader.fieldnames
        return fns is not None and "text" in fns and "subject_uris" in fns

    @staticmethod
    def is_csv_file(path: str) -> bool:
        """return True if the path looks like a CSV file"""

        path_lc = path.lower()
        return path_lc.endswith(".csv") or path_lc.endswith(".csv.gz")


class DocumentFileJSONL(DocumentCorpus):
    """A JSON Lines file as a corpus of documents with subjects"""

    def __init__(self, path: str, subject_index: SubjectIndex, language: str) -> None:
        self.path = path
        self.subject_index = subject_index
        self.language = language

    @property
    def documents(self) -> Iterator[Document]:
        if self.path.endswith(".gz"):
            opener = gzip.open
        else:
            opener = open
        with opener(self.path, mode="rt", encoding="utf-8") as jsonlfile:
            for line in jsonlfile:
                doc = json_to_document(
                    self.path, line, self.subject_index, self.language, True
                )
                if doc is not None:
                    yield doc

    @staticmethod
    def is_jsonl_file(path: str) -> bool:
        """return True if the path looks like a JSONL file"""

        path_lc = path.lower()
        return path_lc.endswith(".jsonl") or path_lc.endswith(".jsonl.gz")


class DocumentList(DocumentCorpus):
    """A document corpus based on a list of other iterable of Document
    objects"""

    def __init__(self, documents):
        self._documents = documents

    @property
    def documents(self):
        yield from self._documents


class TransformingDocumentCorpus(DocumentCorpus):
    """A document corpus that wraps another document corpus but transforms the
    documents using a given transform function"""

    def __init__(self, corpus, transform_fn):
        self._orig_corpus = corpus
        self._transform_fn = transform_fn

    @property
    def documents(self):
        for doc in self._orig_corpus.documents:
            yield self._transform_fn(doc)


class LimitingDocumentCorpus(DocumentCorpus):
    """A document corpus that wraps another document corpus but limits the
    number of documents to a given limit"""

    def __init__(self, corpus, docs_limit):
        self._orig_corpus = corpus
        self.docs_limit = docs_limit

    @property
    def documents(self):
        for doc in islice(self._orig_corpus.documents, self.docs_limit):
            yield doc


1			"""Classes for supporting document corpora"""
2
3			from __future__ import annotations
4
5			import csv
6			import glob
7			import gzip
8			import os.path
9			import re
10			from itertools import islice
11			from typing import TYPE_CHECKING
12
13			import annif.util
14			from annif.exception import OperationFailedException
15
16			from .json import json_file_to_document, json_to_document
17			from .types import Document, DocumentCorpus, SubjectSet
18
19			if TYPE_CHECKING:
20			from collections.abc import Iterator
21
22			from annif.vocab import SubjectIndex
23
24			logger = annif.logger
25
26
27			class DocumentDirectory(DocumentCorpus):
28			"""A directory of files as a full text document corpus"""
29
30			def __init__(
31			self,
32			path: str,
33			subject_index: SubjectIndex \| None = None,
34			language: str \| None = None,
35			require_subjects: bool = False,
36			) -> None:
37			self.path = path
38			self.subject_index = subject_index
39			self.language = language
40			self.require_subjects = require_subjects
41
42			def __iter__(self) -> Iterator[str]:
43			"""Iterate through the directory, yielding file paths with corpus documents."""
44
45			# txt files
46			for filename in sorted(glob.glob(os.path.join(self.path, "*.txt"))):
47			yield filename
48
49			# json files
50			for filename in sorted(glob.glob(os.path.join(self.path, "*.json"))):
51			yield filename
52
53			@staticmethod
54			def _get_subject_filename(filename: str) -> str \| None:
55			tsvfilename = re.sub(r"\.txt$", ".tsv", filename)
56			if os.path.exists(tsvfilename):
57			return tsvfilename
58
59			keyfilename = re.sub(r"\.txt$", ".key", filename)
60			if os.path.exists(keyfilename):
61			return keyfilename
62
63			return None
64
65			def _read_txt_file(self, filename: str) -> Document \| None:
66			with open(filename, errors="replace", encoding="utf-8-sig") as docfile:
67			text = docfile.read()
68			if not self.require_subjects:
69			return Document(text=text, subject_set=None, file_path=filename)
70
71			subjfilename = self._get_subject_filename(filename)
72			if subjfilename is None:
73			# subjects required but not found, skipping this docfile
74			return None
75
76			with open(subjfilename, encoding="utf-8-sig") as subjfile:
77			subjects = SubjectSet.from_string(
78			subjfile.read(), self.subject_index, self.language
79			)
80			return Document(text=text, subject_set=subjects, file_path=filename)
81
82			@property
83			def documents(self) -> Iterator[Document]:
84			for docfilename in self:
85			if docfilename.endswith(".txt"):
86			doc = self._read_txt_file(docfilename)
87			else:
88			doc = json_file_to_document(
89			docfilename,
90			self.subject_index,
91			self.language,
92			self.require_subjects,
93			)
94
95			if doc is not None:
96			yield doc
97
98
99			class DocumentFileTSV(DocumentCorpus):
100			"""A TSV file as a corpus of documents with subjects"""
101
102			def __init__(self, path: str, subject_index: SubjectIndex) -> None:
103			self.path = path
104			self.subject_index = subject_index
105
106			@property
107			def documents(self) -> Iterator[Document]:
108			if self.path.endswith(".gz"):
109			opener = gzip.open
110			else:
111			opener = open
112			with opener(self.path, mode="rt", encoding="utf-8-sig") as tsvfile:
113			for line in tsvfile:
114			yield from self._parse_tsv_line(line)
115
116			def _parse_tsv_line(self, line: str) -> Iterator[Document]:
117			if "\t" in line:
118			text, uris = line.split("\t", maxsplit=1)
119			subject_ids = {
120			self.subject_index.by_uri(annif.util.cleanup_uri(uri))
121			for uri in uris.split()
122			}
123			yield Document(text=text, subject_set=SubjectSet(subject_ids))
124			else:
125			logger.warning('Skipping invalid line (missing tab): "%s"', line.rstrip())
126
127
128			class DocumentFileCSV(DocumentCorpus):
129			"""A CSV file as a corpus of documents with subjects"""
130
131			def __init__(self, path: str, subject_index: SubjectIndex) -> None:
132			self.path = path
133			self.subject_index = subject_index
134
135			@property
136			def documents(self) -> Iterator[Document]:
137			if self.path.endswith(".gz"):
138			opener = gzip.open
139			else:
140			opener = open
141			with opener(self.path, mode="rt", encoding="utf-8-sig") as csvfile:
142			reader = csv.DictReader(csvfile)
143			if not self._check_fields(reader):
144			raise OperationFailedException(
145			f"Cannot parse CSV file {self.path}. "
146			+ "The file must have a header row that defines at least "
147			+ "the columns 'text' and 'subject_uris'."
148			)
149			for row in reader:
150			yield from self._parse_row(row)
151
152			def _parse_row(self, row: dict[str, str]) -> Iterator[Document]:
153			subject_ids = {
154			self.subject_index.by_uri(annif.util.cleanup_uri(uri))
155			for uri in (row["subject_uris"] or "").strip().split()
156			}
157			metadata = {
158			key: val for key, val in row.items() if key not in ("text", "subject_uris")
159			}
160			yield Document(
161			text=(row["text"] or ""),
162			subject_set=SubjectSet(subject_ids),
163			metadata=metadata,
164			)
165
166			def _check_fields(self, reader: csv.DictReader) -> bool:
167			fns = reader.fieldnames
168			return fns is not None and "text" in fns and "subject_uris" in fns
169
170			@staticmethod
171			def is_csv_file(path: str) -> bool:
172			"""return True if the path looks like a CSV file"""
173
174			path_lc = path.lower()
175			return path_lc.endswith(".csv") or path_lc.endswith(".csv.gz")
176
177
178			class DocumentFileJSONL(DocumentCorpus):
179			"""A JSON Lines file as a corpus of documents with subjects"""
180
181			def __init__(self, path: str, subject_index: SubjectIndex, language: str) -> None:
182			self.path = path
183			self.subject_index = subject_index
184			self.language = language
185
186			@property
187			def documents(self) -> Iterator[Document]:
188			if self.path.endswith(".gz"):
189			opener = gzip.open
190			else:
191			opener = open
192			with opener(self.path, mode="rt", encoding="utf-8") as jsonlfile:
193			for line in jsonlfile:
194			doc = json_to_document(
195			self.path, line, self.subject_index, self.language, True
196			)
197			if doc is not None:
198			yield doc
199
200			@staticmethod
201			def is_jsonl_file(path: str) -> bool:
202			"""return True if the path looks like a JSONL file"""
203
204			path_lc = path.lower()
205			return path_lc.endswith(".jsonl") or path_lc.endswith(".jsonl.gz")
206
207
208			class DocumentList(DocumentCorpus):
209			"""A document corpus based on a list of other iterable of Document
210			objects"""
211
212			def __init__(self, documents):
213			self._documents = documents
214
215			@property
216			def documents(self):
217			yield from self._documents
218
219
220			class TransformingDocumentCorpus(DocumentCorpus):
221			"""A document corpus that wraps another document corpus but transforms the
222			documents using a given transform function"""
223
224			def __init__(self, corpus, transform_fn):
225			self._orig_corpus = corpus
226			self._transform_fn = transform_fn
227
228			@property
229			def documents(self):
230			for doc in self._orig_corpus.documents:
231			yield self._transform_fn(doc)
232
233
234			class LimitingDocumentCorpus(DocumentCorpus):
235			"""A document corpus that wraps another document corpus but limits the
236			number of documents to a given limit"""
237
238			def __init__(self, corpus, docs_limit):
239			self._orig_corpus = corpus
240			self.docs_limit = docs_limit
241
242			@property
243			def documents(self):
244			for doc in islice(self._orig_corpus.documents, self.docs_limit):
245			yield doc
246

NatLibFi / Annif

Pull Request — main (#876)

annif.corpus.document.DocumentFileCSV._parse_row() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like