From 529ad3fdb44cb4709bac97a0904a092557ea7009 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Wed, 26 Aug 2026 22:25:51 +0200 Subject: [PATCH 1/2] maintenance: modernize inference code and tests --- .github/workflows/codeql-analysis.yml | 70 ---- .github/workflows/tests.yml | 62 +--- README.rst | 7 - py3langid/langid.py | 478 ++++++++++---------------- pyproject.toml | 17 +- tests/test_langid.py | 152 ++++++-- tests/test_server.py | 172 ++++----- 7 files changed, 389 insertions(+), 569 deletions(-) delete mode 100644 .github/workflows/codeql-analysis.yml diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml deleted file mode 100644 index 6f64e5c..0000000 --- a/.github/workflows/codeql-analysis.yml +++ /dev/null @@ -1,70 +0,0 @@ -# For most projects, this workflow file will not need changing; you simply need -# to commit it to your repository. -# -# You may wish to alter this file to override the set of languages analyzed, -# or to provide custom queries or build logic. -# -# ******** NOTE ******** -# We have attempted to detect the languages in your repository. Please check -# the `language` matrix defined below to confirm you have the correct set of -# supported CodeQL languages. -# -name: "CodeQL" - -on: - push: - branches: [ master ] - pull_request: - # The branches below must be a subset of the branches above - branches: [ master ] - schedule: - - cron: '23 1 * * 1' - -jobs: - analyze: - name: Analyze - runs-on: ubuntu-latest - permissions: - actions: read - contents: read - security-events: write - - strategy: - fail-fast: false - matrix: - language: [ 'python' ] - # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ] - # Learn more about CodeQL language support at https://git.io/codeql-language-support - - steps: - - name: Checkout repository - uses: actions/checkout@v2 - - # Initializes the CodeQL tools for scanning. - - name: Initialize CodeQL - uses: github/codeql-action/init@v1 - with: - languages: ${{ matrix.language }} - # If you wish to specify custom queries, you can do so here or in a config file. - # By default, queries listed here will override any specified in a config file. - # Prefix the list here with "+" to use these queries and those in the config file. - # queries: ./path/to/local/query, your-org/your-repo/queries@main - - # Autobuild attempts to build any compiled languages (C/C++, C#, or Java). - # If this step fails, then you should remove it and run the build manually (see below) - - name: Autobuild - uses: github/codeql-action/autobuild@v1 - - # ℹ️ Command-line programs to run using the OS shell. - # 📚 https://git.io/JvXDl - - # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines - # and modify them (or add more) to build your code if your project - # uses a compiled language - - #- run: | - # make bootstrap - # make release - - - name: Perform CodeQL Analysis - uses: github/codeql-action/analyze@v1 diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index b95b336..89a244e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -1,61 +1,29 @@ -# This workflow will install Python dependencies, run tests and lint with a variety of Python versions -# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions - name: Tests on: push: - branches: [ master ] + branches: [master] pull_request: - branches: [ master ] + branches: [master] jobs: - build: - + test: runs-on: ${{ matrix.os }} strategy: fail-fast: false matrix: os: [ubuntu-latest] - # https://github.com/actions/python-versions/blob/main/versions-manifest.json - python-version: [3.8, 3.9, "3.10", "3.11", "3.12", "3.13-dev"] + python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] include: - # other OS version necessary - - os: macos-latest - python-version: "3.10" - - os: windows-latest - python-version: "3.10" + - os: macos-latest + python-version: "3.12" + - os: windows-latest + python-version: "3.12" steps: - # Python and pip setup - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - - - name: Upgrade pip - run: python -m pip install --upgrade pip - - - name: Get pip cache dir - id: pip-cache - run: | - echo "::set-output name=dir::$(pip cache dir)" - - - name: pip cache - uses: actions/cache@v4 - with: - path: ${{ steps.pip-cache.outputs.dir }} - key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }} - restore-keys: | - ${{ runner.os }}-pip- - - # package setup - - uses: actions/checkout@v4 - - - name: Install dependencies - run: python -m pip install -e "." - - # tests - - name: Test with pytest - run: | - python -m pip install pytest pytest-cov - pytest --cov=./ --cov-report=xml + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: pip install -e ".[dev]" + - run: pytest diff --git a/README.rst b/README.rst index 1bb793f..37ad9de 100644 --- a/README.rst +++ b/README.rst @@ -78,13 +78,6 @@ More options: ('en', 1.0) -Note: the Numpy data type for the feature vector has been changed to optimize for speed. If results are inconsistent, try restoring the original setting: - -.. code-block:: python - - >>> langid.classify(text, datatype='uint32') - - On the command-line ~~~~~~~~~~~~~~~~~~~ diff --git a/py3langid/langid.py b/py3langid/langid.py index 2358ea6..c945c7c 100755 --- a/py3langid/langid.py +++ b/py3langid/langid.py @@ -1,3 +1,4 @@ +#!/usr/bin/env python3 """ This file bundles language identification functions. @@ -14,372 +15,266 @@ import logging import lzma import pickle - from base64 import b64decode from collections import Counter +from http import HTTPStatus from operator import itemgetter from pathlib import Path from urllib.parse import parse_qs import numpy as np - LOGGER = logging.getLogger(__name__) -# model defaults IDENTIFIER = None MODEL_FILE = 'data/model.plzma' -NORM_PROBS = False # Normalize output probabilities. -# NORM_PROBS defaults to False for a small speed increase. It does not -# affect the relative ordering of the predicted classes. It can be -# re-enabled at runtime - see the readme. - -# quantization: faster but less precise -DATATYPE = "uint16" +MODEL_DIR = Path(__file__).parent -def load_model(path=None): - """ - Convenience method to set the global identifier using a model at a - specified path. - - @param path to model - """ - LOGGER.debug('initializing identifier') +def _load_identifier(model_path=None, norm_probs=False, langs=None): + """Load an identifier: external model if given, else the bundled one.""" + identifier = None + if model_path: + try: + identifier = LanguageIdentifier.from_modelpath(model_path, norm_probs=norm_probs) + LOGGER.info("Using external model: %s", model_path) + except OSError as e: + LOGGER.warning("Failed to load %s: %s", model_path, e) + if identifier is None: + identifier = LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=norm_probs) + if langs: + identifier.set_languages(langs) + return identifier + + +def _get_identifier(): + """Return the global identifier, loading the default model if needed.""" global IDENTIFIER - if path is None: - IDENTIFIER = LanguageIdentifier.from_pickled_model(MODEL_FILE) - else: - IDENTIFIER = LanguageIdentifier.from_modelpath(path) + if IDENTIFIER is None: + LOGGER.debug('initializing identifier') + IDENTIFIER = _load_identifier() + return IDENTIFIER def set_languages(langs=None): - """ - Set the language set used by the global identifier. - - @param langs a list of language codes - """ - if IDENTIFIER is None: - load_model() - return IDENTIFIER.set_languages(langs) + """Set the language subset used by the global identifier.""" + return _get_identifier().set_languages(langs) -def classify(instance, datatype=DATATYPE): - """ - Convenience method using a global identifier instance with the default - model included in langid.py. Identifies the language that a string is - written in. - - @param instance a text string. Unicode strings will automatically be utf8-encoded - @returns a tuple of the most likely language and the confidence score - """ - if IDENTIFIER is None: - load_model() - return IDENTIFIER.classify(instance, datatype=datatype) +def classify(instance): + """Classify a text string, returning (language, confidence).""" + return _get_identifier().classify(instance) def rank(instance): - """ - Convenience method using a global identifier instance with the default - model included in langid.py. Ranks all the languages in the model according - to the likelihood that the string is written in each language. - - @param instance a text string. Unicode strings will automatically be utf8-encoded - @returns a list of tuples language and the confidence score, in descending order - """ - if IDENTIFIER is None: - load_model() - return IDENTIFIER.rank(instance) - - -def cl_path(path): - """ - Convenience method using a global identifier instance with the default - model included in langid.py. Identifies the language that the file at `path` is - written in. + """Rank all languages by likelihood, returning [(language, confidence), ...].""" + return _get_identifier().rank(instance) - @param path path to file - @returns a tuple of the most likely language and the confidence score - """ - if IDENTIFIER is None: - load_model() - return IDENTIFIER.cl_path(path) +def _init_worker(model_path, norm_probs, langs): + # spawned Pool workers get a fresh module: rebuild the parent's identifier + global IDENTIFIER + IDENTIFIER = _load_identifier(model_path, norm_probs, langs) -def rank_path(path): - """ - Convenience method using a global identifier instance with the default - model included in langid.py. Ranks all the languages in the model according - to the likelihood that the file at `path` is written in each language. - @param path path to file - @returns a list of tuples language and the confidence score, in descending order - """ - if IDENTIFIER is None: - load_model() - return IDENTIFIER.rank_path(path) +def _process_file(path, dist=False): + with open(path, 'rb') as f: + text = f.read() + return path, (rank(text) if dist else classify(text)) class LanguageIdentifier: - """ - This class implements the actual language identifier. - """ - __slots__ = ['nb_ptc', 'nb_pc', 'nb_numfeats', 'nb_classes', 'tk_nextmove', 'tk_output', - 'norm_probs', '__full_model'] + __slots__ = [ + '_full_model', + '_norm_probs', + 'nb_classes', + 'nb_ptc', + 'tk_nextmove', + 'tk_output', + ] - # new version: speed-up @classmethod - def from_pickled_model(cls, pickled_file, *args, **kwargs): - # load data - filepath = str(Path(__file__).parent / pickled_file) - with lzma.open(filepath) as filehandle: - nb_ptc, nb_pc, nb_classes, tk_nextmove, tk_output = pickle.load(filehandle) - nb_numfeats = len(nb_ptc) // len(nb_pc) - - # reconstruct pc and ptc - nb_pc = np.array(nb_pc) - nb_ptc = np.array(nb_ptc).reshape(nb_numfeats, len(nb_pc)) + def _from_model_data(cls, nb_ptc, _nb_pc, nb_classes, tk_nextmove, tk_output, *args, **kwargs): + n_classes = len(nb_classes) + nb_ptc = np.array(nb_ptc).reshape(len(nb_ptc) // n_classes, n_classes) + # {state: features} dict -> dense list; 256 transitions per DFA state + output_list = [None] * (len(tk_nextmove) // 256) + for s, v in tk_output.items(): + output_list[s] = v + return cls(nb_ptc, nb_classes, tk_nextmove, output_list, *args, **kwargs) - return cls(nb_ptc, nb_pc, nb_numfeats, nb_classes, tk_nextmove, tk_output, *args, **kwargs) + @classmethod + def from_pickled_model(cls, pickled_file, *args, **kwargs): + with lzma.open(MODEL_DIR / pickled_file) as f: + data = pickle.load(f) + return cls._from_model_data(*data, *args, **kwargs) - # legacy methods @classmethod def from_modelstring(cls, string, *args, **kwargs): - # load data - nb_ptc, nb_pc, nb_classes, tk_nextmove, tk_output = pickle.loads(bz2.decompress(b64decode(string))) - nb_numfeats = len(nb_ptc) // len(nb_pc) - - # reconstruct pc and ptc - nb_pc = np.array(nb_pc) - nb_ptc = np.array(nb_ptc).reshape(nb_numfeats, len(nb_pc)) - - return cls(nb_ptc, nb_pc, nb_numfeats, nb_classes, tk_nextmove, tk_output, *args, **kwargs) + data = pickle.loads(bz2.decompress(b64decode(string))) + return cls._from_model_data(*data, *args, **kwargs) @classmethod def from_modelpath(cls, path, *args, **kwargs): with open(path, 'rb') as f: return cls.from_modelstring(f.read(), *args, **kwargs) - def __init__(self, nb_ptc, nb_pc, nb_numfeats, nb_classes, tk_nextmove, tk_output, - norm_probs=NORM_PROBS): + def __init__(self, nb_ptc, nb_classes, tk_nextmove, tk_output, norm_probs=False): self.nb_ptc = nb_ptc - self.nb_pc = nb_pc - self.nb_numfeats = nb_numfeats self.nb_classes = nb_classes self.tk_nextmove = tk_nextmove self.tk_output = tk_output - - def apply_norm_probs(pd): - """ - Renormalize log-probs into a proper distribution (sum 1) - The technique for dealing with underflow is described in - http://jblevins.org/log/log-sum-exp - """ - if norm_probs: - # Ignore overflow when computing the exponential. Large values - # in the exp produce a result of inf, which does not affect - # the correctness of the calculation (as 1/x->0 as x->inf). - # On Linux this does not actually trigger a warning, but on - # Windows this causes a RuntimeWarning, so we explicitly - # suppress it. - with np.errstate(over='ignore'): - # legacy formula, there are possibly better alternatives - pd = 1/np.exp(pd[None,:] - pd[:,None]).sum(1) - return pd - - self.norm_probs = apply_norm_probs - - # Maintain a reference to the full model, in case we change our language set - # multiple times. - self.__full_model = nb_ptc, nb_pc, nb_classes + self._norm_probs = norm_probs + self._full_model = nb_ptc, nb_classes def set_languages(self, langs=None): LOGGER.debug("restricting languages to: %s", langs) - - # Unpack the full original model. This is needed in case the language set - # has been previously trimmed, and the new set is not a subset of the current - # set. - nb_ptc, nb_pc, nb_classes = self.__full_model + nb_ptc, nb_classes = self._full_model if langs is None: - self.nb_classes, self.nb_ptc, self.nb_pc = nb_classes, nb_ptc, nb_pc - + self.nb_classes, self.nb_ptc = nb_classes, nb_ptc else: - # We were passed a restricted set of languages. Trim the arrays accordingly - # to speed up processing. - for lang in langs: - if lang not in nb_classes: - raise ValueError(f"Unknown language code {lang}") - - subset_mask = np.isin(nb_classes, langs) - self.nb_classes = [c for c in nb_classes if c in langs] - self.nb_ptc = nb_ptc[:, subset_mask] - self.nb_pc = nb_pc[subset_mask] - - def instance2fv(self, text, datatype=DATATYPE): - """ - Map an instance into the feature space of the trained model. - - @param datatype NumPy data type (originally uint32) - """ - # convert to binary if it isn't already the case + lang_set = set(langs) + unknown = lang_set - set(nb_classes) + if unknown: + raise ValueError(f"Unknown language code(s): {unknown}") + + indices = [i for i, c in enumerate(nb_classes) if c in lang_set] + self.nb_classes = [nb_classes[i] for i in indices] + self.nb_ptc = nb_ptc[:, indices] + + def _score(self, text): + if isinstance(text, bytes): + # decode so case normalization applies uniformly to str and bytes + try: + text = text.decode('utf8') + except UnicodeDecodeError: + pass if isinstance(text, str): - # fix for surrogates on Windows/NT platforms + if text.isupper(): + text = text.lower() text = text.encode('utf8', errors='surrogatepass') - # Convert the text to a sequence of ascii values and - # Count the number of times we enter each state + # DFA walk state, indexes = 0, [] extend = indexes.extend + nm, out = self.tk_nextmove, self.tk_output for letter in text: - state = self.tk_nextmove[(state << 8) + letter] - extend(self.tk_output.get(state, [])) - - # datatype: consider that less feature counts are going to be needed - arr = np.zeros(self.nb_numfeats, dtype=datatype) - # Update all the productions corresponding to the state - for index, value in Counter(indexes).items(): - arr[index] = value - - return arr - - def nb_classprobs(self, fv): - # compute the partial log-probability of the document given each class - pdc = np.dot(fv, self.nb_ptc) # fv @ self.nb_ptc - # compute the partial log-probability of the document in each class - return pdc + self.nb_pc - - def classify(self, text, datatype=DATATYPE): - """ - Classify an instance. - """ - fv = self.instance2fv(text, datatype=datatype) - probs = self.norm_probs(self.nb_classprobs(fv)) - cl = np.argmax(probs) - return self.nb_classes[cl], probs[cl] + state = nm[(state << 8) + letter] + v = out[state] + if v: + extend(v) + + if indexes: + feat_counts = Counter(indexes) + idx = np.fromiter(feat_counts.keys(), dtype=np.intp, count=len(feat_counts)) + counts = np.fromiter(feat_counts.values(), dtype=np.float32, count=len(feat_counts)) + probs = counts @ self.nb_ptc[idx] + else: + # no features: minimal confidence (softmax turns this into uniform) + fill = 0.0 if self._norm_probs else -np.inf + probs = np.full(len(self.nb_classes), fill, dtype=np.float32) - def rank(self, text): - """ - Return a list of languages in order of likelihood. - """ - fv = self.instance2fv(text) - probs = self.norm_probs(self.nb_classprobs(fv)) - return sorted(zip(self.nb_classes, probs), key=itemgetter(1), reverse=True) - - def cl_path(self, path): - """ - Classify a file at a given path - """ - with open(path, 'rb') as f: - retval = self.classify(f.read()) - return path, retval + if self._norm_probs: + e = np.exp(probs - probs.max()) + probs = e / e.sum() - def rank_path(self, path): - """ - Class ranking for a file at a given path - """ - with open(path, 'rb') as f: - retval = self.rank(f.read()) - return path, retval + return probs + + def classify(self, text): + probs = self._score(text) + cl = probs.argmax() + return self.nb_classes[cl], float(probs[cl]) + + def rank(self, text): + probs = self._score(text) + return sorted( + ((lang, float(p)) for lang, p in zip(self.nb_classes, probs)), + key=itemgetter(1), reverse=True, + ) -class NumpyEncoder(json.JSONEncoder): - """ Custom encoder for numpy data types """ - def default(self, o): - if isinstance(o, np.float32): - return float(o) # Convert float32 to native float - if isinstance(o, np.ndarray): - return o.tolist() # Convert arrays to list - return json.JSONEncoder.default(self, o) +def _detect(data): + lang, conf = classify(data) + return {'language': lang, 'confidence': conf} -METHODS = { - 'detect': lambda data: {'language': classify(data)[0], 'confidence': classify(data)[1]}, - 'rank': lambda data: rank(data) -} +_ROUTES = {'detect': _detect, 'rank': rank} def application(environ, start_response): - """ - WSGI-compatible langid web service. - """ - from wsgiref.util import shift_path_info - try: - path = shift_path_info(environ) - except IndexError: - # Catch shift_path_info's failure to handle empty paths properly - path = '' - - if path not in METHODS: + """WSGI-compatible langid web service.""" + path = environ.get('PATH_INFO', '').strip('/').partition('/')[0] + handler = _ROUTES.get(path) + if handler is None: return _return_response(start_response, 404, None, 'Not found') + method = environ['REQUEST_METHOD'] + if method not in ('GET', 'POST', 'PUT'): + return _return_response(start_response, 405, None, f'{method} not allowed') + data = _get_data(environ) if data is None: - if environ['REQUEST_METHOD'] == 'GET' and 'QUERY_STRING' not in environ: - return _return_response(start_response, 400, None, 'Missing query string') - return _return_response(start_response, 405, None, f"{environ['REQUEST_METHOD']} not allowed") + return _return_response(start_response, 400, None, 'No data provided') - response_data = METHODS[path](data) - return _return_response(start_response, 200, response_data, None) + return _return_response(start_response, 200, handler(data), None) def _get_data(environ): - if environ['REQUEST_METHOD'] in ['PUT', 'POST']: - data = environ['wsgi.input'].read(int(environ['CONTENT_LENGTH'])) - if environ['REQUEST_METHOD'] == 'POST': + method = environ['REQUEST_METHOD'] + if method in ('PUT', 'POST'): + try: + length = int(environ.get('CONTENT_LENGTH', 0)) + except ValueError: + return None + if length <= 0: + return None + data = environ['wsgi.input'].read(length) + if method == 'POST': try: - data = parse_qs(data)['q'][0] + data = parse_qs(data)[b'q'][0] except KeyError: pass return data - if environ['REQUEST_METHOD'] == 'GET': + if method == 'GET': try: - return parse_qs(environ['QUERY_STRING'])['q'][0] + return parse_qs(environ.get('QUERY_STRING', ''))['q'][0] except KeyError: - pass + return None return None -STATUS_MESSAGES = { - 200: "OK", - 404: "Not Found", - 405: "Method Not Allowed" -} - - def _return_response(start_response, status_code, response_data, response_details): - status = f"{status_code} {STATUS_MESSAGES.get(status_code, 'Unknown Status')}" + status = HTTPStatus(status_code) response = { 'responseData': response_data, 'responseStatus': status_code, 'responseDetails': response_details, } - headers = [('Content-type', 'text/javascript; charset=utf-8')] - start_response(status, headers) - return [json.dumps(response, cls=NumpyEncoder).encode('utf-8')] + headers = [('Content-type', 'application/json; charset=utf-8')] + start_response(f"{status.value} {status.phrase}", headers) + return [json.dumps(response).encode('utf-8')] def main(): - # lazy imports import argparse import sys - # parse arguments parser = argparse.ArgumentParser() - parser.add_argument('-s', '--serve', action='store_true', default=False, dest='serve', help='launch web service') - parser.add_argument('--host', default=None, dest='host', help='host/ip to bind to') - parser.add_argument('--port', default=9008, dest='port', help='port to listen on') + parser.add_argument('-s', '--serve', action='store_true', help='launch web service') + parser.add_argument('--host', help='host/ip to bind to') + parser.add_argument('--port', default=9008, type=int, help='port to listen on') parser.add_argument('-v', action='count', dest='verbosity', help='increase verbosity (repeat for greater effect)') parser.add_argument('-m', dest='model', help='load model from file') - parser.add_argument('-l', '--langs', dest='langs', help='comma-separated set of target ISO639 language codes (e.g en,de)') - parser.add_argument('-r', '--remote', action="store_true", default=False, help='auto-detect IP address for remote access') - parser.add_argument('-b', '--batch', action="store_true", default=False, help='specify a list of files on the command line') - parser.add_argument('-d', '--dist', action='store_true', default=False, help='show full distribution over languages') + parser.add_argument('-l', '--langs', help='comma-separated set of target ISO639 language codes (e.g en,de)') + parser.add_argument('-r', '--remote', action='store_true', help='auto-detect IP address for remote access') + parser.add_argument('-b', '--batch', action='store_true', help='specify a list of files on the command line') + parser.add_argument('-d', '--dist', action='store_true', help='show full distribution over languages') parser.add_argument('-u', '--url', help='langid of URL') - parser.add_argument('--line', action="store_true", default=False, help='process pipes line-by-line rather than as a document') - parser.add_argument('-n', '--normalize', action='store_true', default=False, help='normalize confidence scores to probability values') + parser.add_argument('--line', action='store_true', help='process pipes line-by-line rather than as a document') + parser.add_argument('-n', '--normalize', action='store_true', help='normalize confidence scores to probability values') options = parser.parse_args() if options.verbosity: @@ -390,29 +285,12 @@ def main(): if options.batch and options.serve: parser.error("cannot specify both batch and serve at the same time") - # unpack a model global IDENTIFIER - if options.model: - try: - IDENTIFIER = LanguageIdentifier.from_modelpath(options.model, norm_probs=options.normalize) - LOGGER.info("Using external model: %s", options.model) - except IOError as e: - LOGGER.warning("Failed to load %s: %s", options.model, e) - - if IDENTIFIER is None: - IDENTIFIER = LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=options.normalize) - LOGGER.info("Using internal model") - - if options.langs: - langs = options.langs.split(",") - IDENTIFIER.set_languages(langs) + langs = options.langs.split(",") if options.langs else None + IDENTIFIER = _load_identifier(options.model, options.normalize, langs) - def _process(text): - """ - Set up a local function to do output, configured according to our settings. - """ - return IDENTIFIER.rank(text) if options.dist else IDENTIFIER.classify(text) + _process = IDENTIFIER.rank if options.dist else IDENTIFIER.classify if options.url: from urllib.request import urlopen @@ -425,48 +303,45 @@ def _process(text): import socket from wsgiref.simple_server import make_server - # from http://stackoverflow.com/questions/166506/finding-local-ip-addresses-in-python if options.remote and options.host is None: - # resolve the external ip address - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("google.com", 80)) - hostname = s.getsockname()[0] + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + s.connect(("google.com", 80)) + hostname = s.getsockname()[0] elif options.host is None: - # resolve the local hostname hostname = socket.gethostbyname(socket.gethostname()) else: hostname = options.host - print(f"Listening on {hostname}:%{options.port}") + print(f"Listening on {hostname}:{options.port}") print("Press Ctrl+C to exit") - httpd = make_server(hostname, int(options.port), application) + httpd = make_server(hostname, options.port, application) try: httpd.serve_forever() except KeyboardInterrupt: pass elif options.batch: - # Start in batch mode - interpret input as paths rather than content - # to classify. import csv + from functools import partial from multiprocessing import Pool def generate_paths(): for line in sys.stdin: - path = line.strip() - if path and Path.is_file(path): - yield path + p = line.strip() + if p and Path(p).is_file(): + yield p writer = csv.writer(sys.stdout) - with Pool() as pool: + with Pool(initializer=_init_worker, + initargs=(options.model, options.normalize, langs)) as pool: if options.dist: writer.writerow(['path'] + IDENTIFIER.nb_classes) - for path, ranking in pool.imap_unordered(rank_path, generate_paths()): + for path, ranking in pool.imap_unordered(partial(_process_file, dist=True), generate_paths()): ranking = dict(ranking) row = [path] + [ranking[c] for c in IDENTIFIER.nb_classes] writer.writerow(row) else: - for path, (lang, conf) in pool.imap_unordered(cl_path, generate_paths()): + for path, (lang, conf) in pool.imap_unordered(_process_file, generate_paths()): writer.writerow((path, lang, conf)) else: if sys.stdin.isatty(): @@ -475,8 +350,7 @@ def generate_paths(): try: print(">>>", end=' ') text = input() - except Exception as e: - print(e) + except (KeyboardInterrupt, EOFError): break print(_process(text)) else: diff --git a/pyproject.toml b/pyproject.toml index ed721c1..882fc6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,12 +7,12 @@ build-backend = "setuptools.build_meta" name = "py3langid" description = "Fork of the language identification tool langid.py, featuring a modernized codebase and faster execution times." readme = "README.rst" -license = { text = "BSD" } +license = "BSD-3-Clause" dynamic = ["version"] -requires-python = ">=3.8" +requires-python = ">=3.10" authors = [ {name = "Marco Lui"}, - {name = "Adrien Barbaresi", email = "barbaresi@bbaw.de"} + {name = "Adrien Barbaresi"} ] keywords=[ "language detection", @@ -21,31 +21,26 @@ keywords=[ "langid.py" ] classifiers = [ - # As from http://pypi.python.org/pypi?%3Aaction=list_classifiers - 'Development Status :: 5 - Production/Stable', - #'Development Status :: 6 - Mature', + "Development Status :: 5 - Production/Stable", "Environment :: Console", "Intended Audience :: Developers", "Intended Audience :: Information Technology", "Intended Audience :: Science/Research", - "License :: OSI Approved :: BSD License", "Operating System :: MacOS :: MacOS X", "Operating System :: Microsoft :: Windows", "Operating System :: POSIX :: Linux", "Programming Language :: Python", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Topic :: Scientific/Engineering :: Artificial Intelligence", "Topic :: Text Processing :: Linguistic", ] dependencies = [ - "numpy >= 2.0.0 ; python_version >= '3.9'", - "numpy >= 1.24.3 ; python_version == '3.8'", + "numpy >= 2.0.0", ] # https://setuptools.pypa.io/en/latest/userguide/pyproject_config.html diff --git a/tests/test_langid.py b/tests/test_langid.py index 2d6a2bc..3dc4a50 100644 --- a/tests/test_langid.py +++ b/tests/test_langid.py @@ -1,44 +1,107 @@ +import csv import subprocess import sys - -from io import StringIO from pathlib import Path +import pytest + import py3langid as langid -from py3langid.langid import LanguageIdentifier, MODEL_FILE - - -def test_langid(): - '''Test if the language detection functions work''' - # basic classification - text = b'This text is in English.' - assert langid.classify(text)[0] == 'en' - assert langid.rank(text)[0][0] == 'en' - text = 'This text is in English.' - assert langid.classify(text)[0] == 'en' - assert langid.rank(text)[0][0] == 'en' - text = 'Test Unicode sur du texte en français' - assert langid.classify(text)[0] == 'fr' - assert langid.rank(text)[0][0] == 'fr' - # other datatype - assert langid.classify(text)[1] != langid.classify(text, datatype='uint32')[1] - # normalization of probabilities - identifier = LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=True) - _, normed_prob = identifier.classify(text) - assert 0 <= normed_prob <= 1 - # probability not equal to 1 - _, normed_prob = identifier.classify('This potrebbe essere a test.') - normed_prob == 0.8942321 - # not normalized - identifier = LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=False) - _, prob = identifier.classify(text) +from py3langid.langid import MODEL_FILE, LanguageIdentifier + + +@pytest.fixture +def identifier(): + return LanguageIdentifier.from_pickled_model(MODEL_FILE) + + +@pytest.fixture +def norm_identifier(): + return LanguageIdentifier.from_pickled_model(MODEL_FILE, norm_probs=True) + + +@pytest.mark.parametrize('text,expected', [ + (b'This text is in English.', 'en'), + ('This text is in English.', 'en'), + ('Test Unicode sur du texte en français', 'fr'), +]) +def test_classify_and_rank(text, expected): + assert langid.classify(text)[0] == expected + assert langid.rank(text)[0][0] == expected + + +def test_norm_probs(norm_identifier): + _, prob = norm_identifier.classify('Test Unicode sur du texte en français') + assert 0 <= prob <= 1 + + +def test_unnormalized(identifier): + _, prob = identifier.classify('Test Unicode sur du texte en français') assert prob < 0 - # subset of target languages + + +def test_language_subset(identifier): identifier.set_languages(['de', 'en', 'fr']) assert identifier.classify('这样不好')[0] != 'zh' +def test_allcaps_lowering(): + '''All-caps text should be lowered before classification''' + assert langid.classify('CECI EST UN TEST EN FRANÇAIS')[0] == 'fr' + assert langid.classify('DIES IST EIN DEUTSCHER TEXT')[0] == 'de' + assert langid.classify('ЭТО РУССКИЙ ТЕКСТ ДЛЯ ТЕСТА')[0] == 'ru' + # title case and mixed case are not lowered + assert langid.classify('This is normal English text')[0] == 'en' + assert langid.classify('NASA launched a SpaceX rocket')[0] == 'en' + + +def test_bytes_str_parity(): + '''bytes and str input give identical results, all-caps included''' + text = 'CECI EST UN TEST EN FRANÇAIS' + assert langid.classify(text) == langid.classify(text.encode('utf8')) + assert langid.rank(text) == langid.rank(text.encode('utf8')) + + +def test_empty_and_short(): + '''Feature-less input scores -inf, short input does not crash''' + for empty in ('', b'', '12345'): + lang, score = langid.classify(empty) + assert isinstance(lang, str) + assert score == float('-inf') + lang, score = langid.classify('a') + assert isinstance(lang, str) + + +def test_norm_probs_empty(norm_identifier): + '''norm_probs=True on empty input returns uniform distribution''' + _, prob = norm_identifier.classify('') + assert abs(prob - 1.0 / len(norm_identifier.nb_classes)) < 1e-6 + + +def test_rank_sorted(identifier): + '''rank() returns all languages sorted by descending score''' + ranking = identifier.rank('Test Unicode sur du texte en français') + assert ranking[0][0] == 'fr' + scores = [s for _, s in ranking] + assert all(isinstance(s, float) for s in scores) + assert scores == sorted(scores, reverse=True) + assert len(ranking) == len(identifier.nb_classes) + + +def test_set_languages_error(identifier): + '''set_languages raises on unknown codes''' + with pytest.raises(ValueError, match="Unknown language code"): + identifier.set_languages(['xx_invalid']) + + +def test_set_languages_reset(identifier): + '''set_languages(None) restores the full model''' + full = len(identifier.nb_classes) + identifier.set_languages(['en', 'fr']) + assert len(identifier.nb_classes) == 2 + identifier.set_languages(None) + assert len(identifier.nb_classes) == full + def test_redirection(): '''Test if STDIN redirection works''' @@ -47,10 +110,37 @@ def test_redirection(): readme_path = str(thisdir.parent / 'README.rst') with open(readme_path, 'rb') as f: readme = f.read() - result = subprocess.check_output(['python3', langid_path, '-n'], input=readme) + result = subprocess.check_output([sys.executable, langid_path, '-n'], input=readme) assert b'en' in result and b'1.0' in result +def test_cli_batch(tmp_path): + '''Batch mode classifies files via the multiprocessing pool''' + en = tmp_path / 'en.txt' + en.write_bytes(b'This is an English text for testing purposes.') + fr = tmp_path / 'fr.txt' + fr.write_text('Ceci est un texte en français pour les tests.', encoding='utf8') + paths = f'{en}\n{fr}\n'.encode() + out = subprocess.check_output(['langid', '-b'], input=paths).decode() + results = {row[0]: row[1] for row in csv.reader(out.strip().splitlines())} + assert results[str(en)] == 'en' and results[str(fr)] == 'fr' + + +def test_cli_external_model(tmp_path): + '''-m loads a model in the modelstring format (b64 + bz2 pickle)''' + import bz2 + import lzma + from base64 import b64encode + + from py3langid.langid import MODEL_DIR + with lzma.open(MODEL_DIR / MODEL_FILE) as f: + raw = f.read() + model_path = tmp_path / 'external.model' + model_path.write_bytes(b64encode(bz2.compress(raw, compresslevel=1))) + result = subprocess.check_output(['langid', '-n', '-m', str(model_path)], + input=b'This should be enough text.') + assert b'en' in result and b'1.0' in result + def test_cli(): '''Test console scripts entry point''' diff --git a/tests/test_server.py b/tests/test_server.py index 81bca3b..6e20c9a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,5 +1,5 @@ import json - +from io import BytesIO from unittest.mock import MagicMock import pytest @@ -7,103 +7,73 @@ from py3langid.langid import application -@pytest.fixture -def mock_start_response(): - return MagicMock() - -def test_detect_put(mock_start_response): - environ = { - 'REQUEST_METHOD': 'PUT', - 'CONTENT_LENGTH': 10, - 'wsgi.input': MagicMock(read=lambda x: b'This is a test'), - 'PATH_INFO': '/detect' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData']['language'] == 'en' - -def test_detect_get(mock_start_response): - environ = { - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'q=This+is+a+test', - 'PATH_INFO': '/detect' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData']['language'] == 'en' - -def test_detect_post(mock_start_response): - environ = { - 'REQUEST_METHOD': 'POST', - 'CONTENT_LENGTH': 10, - 'wsgi.input': MagicMock(read=lambda x: b'q=Hello+World'), - 'PATH_INFO': '/detect' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData']['language'] == 'en' - -def test_rank_put(mock_start_response): - environ = { - 'REQUEST_METHOD': 'PUT', - 'CONTENT_LENGTH': 10, - 'wsgi.input': MagicMock(read=lambda x: b'Hello World'), - 'PATH_INFO': '/rank' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData'] is not None - -def test_rank_get(mock_start_response): - environ = { - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'q=Hello+World', - 'PATH_INFO': '/rank' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData'] is not None - -def test_rank_post(mock_start_response): - environ = { - 'REQUEST_METHOD': 'POST', - 'CONTENT_LENGTH': 10, - 'wsgi.input': MagicMock(read=lambda x: b'q=Hello+World'), - 'PATH_INFO': '/rank' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '200 OK' - assert json.loads(response[0].decode('utf-8'))['responseData'] is not None - -def test_invalid_method(mock_start_response): - environ = { - 'REQUEST_METHOD': 'DELETE', - 'PATH_INFO': '/detect' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '405 Method Not Allowed' - -def test_invalid_path(mock_start_response): - environ = { - 'REQUEST_METHOD': 'GET', - 'PATH_INFO': '/invalid' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '404 Not Found' - -def test_empty_path(mock_start_response): - environ = { - 'REQUEST_METHOD': 'GET', - 'PATH_INFO': '' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '404 Not Found' - -def test_no_query_string(mock_start_response): - environ = { - 'REQUEST_METHOD': 'GET', - 'PATH_INFO': '/detect' - } - response = application(environ, mock_start_response) - assert mock_start_response.call_args[0][0] == '400 Unknown Status' - assert json.loads(response[0].decode('utf-8'))['responseData'] is None +def _request(path, method='GET', body=None, query=None, content_length='auto'): + """Run a WSGI request and return (status, payload, headers).""" + start_response = MagicMock() + environ = {'REQUEST_METHOD': method, 'PATH_INFO': path} + if query is not None: + environ['QUERY_STRING'] = query + if body is not None: + environ['wsgi.input'] = BytesIO(body) + if content_length == 'auto': + environ['CONTENT_LENGTH'] = str(len(body)) + elif content_length is not None: + environ['CONTENT_LENGTH'] = content_length + response = application(environ, start_response) + status = start_response.call_args[0][0] + headers = dict(start_response.call_args[0][1]) + payload = json.loads(response[0].decode('utf-8')) + return status, payload, headers + + +@pytest.mark.parametrize('path', ['/detect', '/rank']) +@pytest.mark.parametrize('method,kwargs', [ + ('GET', {'query': 'q=This+is+a+test'}), + ('POST', {'body': b'q=This+is+a+test'}), + ('PUT', {'body': b'This is a test'}), +]) +def test_ok(path, method, kwargs): + status, payload, _ = _request(path, method, **kwargs) + assert status == '200 OK' + data = payload['responseData'] + if path == '/detect': + assert data['language'] == 'en' + else: + assert data[0][0] == 'en' + + +def test_invalid_method(): + status, _, _ = _request('/detect', 'DELETE') + assert status == '405 Method Not Allowed' + + +@pytest.mark.parametrize('path', ['/invalid', '']) +def test_invalid_path(path): + status, _, _ = _request(path) + assert status == '404 Not Found' + + +def test_no_query_string(): + status, payload, _ = _request('/detect') + assert status == '400 Bad Request' + assert payload['responseData'] is None + + +def test_get_missing_q_param(): + status, _, _ = _request('/detect', query='x=hello') + assert status == '400 Bad Request' + + +def test_content_type(): + _, _, headers = _request('/detect', query='q=test') + assert headers['Content-type'] == 'application/json; charset=utf-8' + + +def test_missing_content_length(): + status, _, _ = _request('/detect', 'POST', body=b'q=test', content_length=None) + assert status == '400 Bad Request' + + +def test_invalid_content_length(): + status, _, _ = _request('/detect', 'POST', body=b'q=test', content_length='abc') + assert status == '400 Bad Request' From abb7868a898a2aaa9733e03f6f20f0a7a582761b Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Thu, 27 Aug 2026 17:52:40 +0200 Subject: [PATCH 2/2] fix windows issue --- py3langid/langid.py | 2 +- tests/test_langid.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/py3langid/langid.py b/py3langid/langid.py index c945c7c..76224b1 100755 --- a/py3langid/langid.py +++ b/py3langid/langid.py @@ -331,7 +331,7 @@ def generate_paths(): if p and Path(p).is_file(): yield p - writer = csv.writer(sys.stdout) + writer = csv.writer(sys.stdout, lineterminator='\n') with Pool(initializer=_init_worker, initargs=(options.model, options.normalize, langs)) as pool: if options.dist: diff --git a/tests/test_langid.py b/tests/test_langid.py index 3dc4a50..d471200 100644 --- a/tests/test_langid.py +++ b/tests/test_langid.py @@ -122,7 +122,7 @@ def test_cli_batch(tmp_path): fr.write_text('Ceci est un texte en français pour les tests.', encoding='utf8') paths = f'{en}\n{fr}\n'.encode() out = subprocess.check_output(['langid', '-b'], input=paths).decode() - results = {row[0]: row[1] for row in csv.reader(out.strip().splitlines())} + results = {row[0]: row[1] for row in csv.reader(out.strip().splitlines()) if row} assert results[str(en)] == 'en' and results[str(fr)] == 'fr'