Text Language Detector

Text Language Detector

Problem Statement

When working with multilingual datasets — customer support tickets, scraped web content, social media posts, or user-generated reviews — you often need to know what language each text is written in before you can route, translate, or analyze it. Manually tagging thousands of rows is impractical, and naive heuristics (like checking for accented characters) fail badly on short or mixed-language text.

This script takes a CSV file, detects the language of a specified text column using the langdetect library, and writes a new CSV with two added columns: the ISO 639-1 language code and the detection confidence score. It is designed to be dropped into a data pipeline as a single, self-contained command.

How It Works

The script follows a simple read → transform → write pipeline. It uses Python's built-in csv.DictReader/DictWriter so it streams row-by-row and doesn't hold the entire file in memory — important for large datasets. For each row, it extracts the target column, calls langdetect.detect_langs() which returns a ranked list of Language objects (each with a lang code and prob probability), and writes back the original row plus detected_lang and detected_confidence columns.

Key design decisions: fail-soft per row (a single unparseable cell shouldn't kill the whole job — it's logged and marked unknown), configurable limits on text length (very long strings slow detection down without improving accuracy, so we truncate), and deterministic seeding via langdetect.DetectorFactory.seed so results are reproducible across runs. Errors on the file level (missing file, missing column) are raised as SystemExit with clear messages so the script plays nicely in shell pipelines.

Full Script

#!/usr/bin/env python3
"""
Text Language Detector

Reads a CSV file, detects the language of a specified column using the
`langdetect` library, and writes a new CSV with two extra columns:
`detected_lang` (ISO 639-1 code) and `detected_confidence` (0.0–1.0).

Usage:
    python3 lang_detect.py --input texts.csv --col content --output with_lang.csv
"""

import argparse
import csv
import logging
import os
import sys
from pathlib import Path
from typing import Iterator

from langdetect import DetectorFactory, detect_langs
from langdetect.lang_detect_exception import LangDetectException

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_MAX_CHARS = int(os.environ.get("LANGDETECT_MAX_CHARS", "1000"))
DEFAULT_MIN_CHARS = int(os.environ.get("LANGDETECT_MIN_CHARS", "3"))
DEFAULT_DELIMITER = os.environ.get("LANGDETECT_DELIMITER", ",")
LOG_LEVEL = os.environ.get("LANGDETECT_LOG_LEVEL", "INFO").upper()

# Deterministic output: langdetect uses random sampling internally.
DetectorFactory.seed = 0

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("lang_detect")


def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Detect the language of a text column in a CSV file."
    )
    parser.add_argument("--input", required=True, help="Path to input CSV file.")
    parser.add_argument("--col", required=True, help="Name of the text column to analyze.")
    parser.add_argument("--output", required=True, help="Path to output CSV file.")
    parser.add_argument(
        "--delimiter",
        default=DEFAULT_DELIMITER,
        help=f"CSV delimiter (default: {DEFAULT_DELIMITER!r}).",
    )
    parser.add_argument(
        "--max-chars",
        type=int,
        default=DEFAULT_MAX_CHARS,
        help="Truncate text to this many characters before detection.",
    )
    parser.add_argument(
        "--min-chars",
        type=int,
        default=DEFAULT_MIN_CHARS,
        help="Text shorter than this is marked 'unknown' without detection.",
    )
    return parser.parse_args()


def iter_rows(path: Path, delimiter: str) -> Iterator[dict[str, str]]:
    """Yield rows from a CSV file as dictionaries."""
    with path.open("r", encoding="utf-8", newline="") as fh:
        reader = csv.DictReader(fh, delimiter=delimiter)
        if reader.fieldnames is None:
            raise ValueError(f"Input file {path} appears to be empty.")
        yield from reader


def detect_one(text: str, min_chars: int, max_chars: int) -> tuple[str, float]:
    """
    Detect the language of a single string.

    Returns (lang_code, confidence). Returns ("unknown", 0.0) when the text
    is too short, empty, or when langdetect cannot make a determination.
    """
    if text is None:
        return "unknown", 0.0

    cleaned = text.strip()
    if len(cleaned) < min_chars:
        return "unknown", 0.0

    # Truncate very long strings — detection accuracy plateaus quickly and
    # long inputs slow down the N-gram scoring considerably.
    sample = cleaned[:max_chars]

    try:
        # detect_langs returns a list of Language objects sorted by probability.
        candidates = detect_langs(sample)
    except LangDetectException as exc:
        log.debug("Detection failed for %r: %s", sample[:40], exc)
        return "unknown", 0.0

    if not candidates:
        return "unknown", 0.0

    top = candidates[0]
    return top.lang, float(top.prob)


def process_file(
    input_path: Path,
    output_path: Path,
    column: str,
    delimiter: str,
    min_chars: int,
    max_chars: int,
) -> int:
    """Process the CSV, writing results. Returns the number of rows written."""
    rows = iter_rows(input_path, delimiter)

    # Peek at the first row to validate the column exists and to obtain headers.
    try:
        first = next(rows)
    except StopIteration:
        raise ValueError(f"Input file {input_path} contains no data rows.")

    if column not in first:
        available = ", ".join(first.keys())
        raise ValueError(
            f"Column {column!r} not found in {input_path}. Available: {available}"
        )

    fieldnames = list(first.keys()) + ["detected_lang", "detected_confidence"]

    output_path.parent.mkdir(parents=True, exist_ok=True)
    written = 0
    unknown = 0

    with output_path.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(fh, fieldnames=fieldnames, delimiter=delimiter)
        writer.writeheader()

        # Chain the first row back onto the iterator so we process everything.
        for row in _chain(first, rows):
            lang, conf = detect_one(row.get(column, ""), min_chars, max_chars)
            if lang == "unknown":
                unknown += 1
            row["detected_lang"] = lang
            row["detected_confidence"] = f"{conf:.4f}"
            writer.writerow(row)
            written += 1

    log.info(
        "Processed %d rows (%d unknown) -> %s", written, unknown, output_path
    )
    return written


def _chain(first: dict[str, str], rest: Iterator[dict[str, str]]) -> Iterator[dict[str, str]]:
    """Yield `first` followed by everything in `rest`."""
    yield first
    yield from rest


def main() -> int:
    args = parse_args()

    input_path = Path(args.input).expanduser()
    output_path = Path(args.output).expanduser()

    if not input_path.is_file():
        log.error("Input file does not exist: %s", input_path)
        return 2

    if input_path.resolve() == output_path.resolve():
        log.error("Input and output paths must differ.")
        return 2

    try:
        process_file(
            input_path=input_path,
            output_path=output_path,
            column=args.col,
            delimiter=args.delimiter,
            min_chars=args.min_chars,
            max_chars=args.max_chars,
        )
    except (ValueError, OSError) as exc:
        log.error("Processing failed: %s", exc)
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

The --input, --col, and --output arguments are required and have no environment-variable fallbacks — they are always explicit.

Usage Examples

Basic run — detect the content column and write results:

$ python3 lang_detect.py --input texts.csv --col content --output with_lang.csv
2024-06-01 10:12:03 [INFO] Processed 1500 rows (23 unknown) -> with_lang.csv

Output file gets two new columns:

id,content,detected_lang,detected_confidence
1,"Bonjour tout le monde",fr,0.9997
2,"Hello world",en,0.9996
3,"Hola amigo",es,0.9995
4,"",unknown,0.0000

Tab-separated file with a stricter min length:

$ python3 lang_detect.py --input tickets.tsv --col body \
    --delimiter $'\t' --min-chars 10 --output tickets_lang.tsv

Debugging a run where many rows are unknown:

$ LANGDETECT_LOG_LEVEL=DEBUG python3 lang_detect.py \
    --input posts.csv --col text --output posts_lang.csv
2024-06-01 10:15:22 [DEBUG] Detection failed for '...': No features in text.

Error Handling

• Missing input file → exits with code 2 and a clear log message.
• Input file is empty or has no data rows → ValueError is caught in main() and returns exit code 1.
• Column name not found → error message lists all available columns so you can fix the typo immediately.
• Input equals output path → refuses to run to avoid clobbering the source.
• Per-cell detection failures (LangDetectException, e.g. "No features in text") → the row is written with detected_lang="unknown" and logged at DEBUG level; the job continues.
• Cells shorter than --min-chars → short-circuited to unknown without calling langdetect.
• Unicode / encoding issues → input and output are opened with encoding="utf-8"; if your source is Latin-1, pre-convert it (e.g. with iconv) before running.

Exit codes: 0 success, 1 processing error, 2 usage/argument error.

Extending Ideas

• Parallelize with multiprocessing: langdetect is CPU-bound; wrap detect_one in a concurrent.futures.ProcessPoolExecutor and map over chunks of rows for a 4–8× speedup on large files.
• Add a --filter-lang flag: emit only rows whose detected language matches a target (e.g. --filter-lang en) to build a clean monolingual corpus in one pass.
• Swap in fasttext or pycld3: both are faster and often more accurate on short text. The detect_one interface is deliberately narrow so you can swap the backend without touching the CSV plumbing.
• Stream from stdin / to stdout: replace Path inputs with sys.stdin/sys.stdout when --input - or --output - is passed, so the tool composes cleanly in shell pipelines (cat big.csv | lang_detect.py ... | gzip > out.csv.gz).