laya_core/

lang.rs

//! Dependency-free script and language detection used to route between checkpoints.
//!
//! Routing needs one decision: is this English Latin text, or something the
//! English checkpoint cannot read? Script detection is exact; the Latin-script
//! language guess is an explicitly best-effort stopword and diacritic heuristic.
//!
//! Two CPython behaviours are reproduced deliberately, because both are
//! observable in the returned `script_profile`:
//!
//! * `str.isalpha()` excludes combining marks that Rust's `char::is_alphabetic`
//!   accepts (see [`crate::pycompat::py_isalpha`]).
//! * `max()` over a dict breaks ties by insertion order, so the counts here are
//!   kept in first-seen order rather than in a hash map.

use crate::lang_tables::{NON_EN_DIACRITICS, SCRIPT_RANGES, STOPWORDS};
use crate::pycompat::{py_isalpha, py_round, py_words_lowercase};
use serde::Serialize;
use serde_json::Value;

const MAX_DETECTION_CHARS: usize = 4000;
const MAX_DEPTH: usize = 6;

/// Detection result for a state.
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Detection {
    pub script: String,
    /// Fraction of alphabetic characters per script, in upstream's key order.
    ///
    /// Kept as ordered pairs rather than a map so the key order is preserved,
    /// and serialized as a JSON object to match upstream's dictionary.
    #[serde(serialize_with = "serialize_profile")]
    pub script_profile: Vec<(String, f64)>,
    pub language: Option<String>,
    pub is_english: bool,
    pub non_latin_fraction: f64,
}

fn serialize_profile<S>(
    profile: &[(String, f64)],
    serializer: S,
) -> std::result::Result<S::Ok, S::Error>
where
    S: serde::Serializer,
{
    use serde::ser::SerializeMap;
    let mut map = serializer.serialize_map(Some(profile.len()))?;
    for (name, fraction) in profile {
        map.serialize_entry(name, fraction)?;
    }
    map.end()
}

impl Detection {
    fn latin_fraction(&self) -> f64 {
        self.script_profile
            .iter()
            .find(|(name, _)| name == "latin")
            .map_or(0.0, |(_, value)| *value)
    }
}

/// Collect the string leaves of a state, so detection sees real content.
fn iter_text<'a>(state: &'a Value, depth: usize, out: &mut Vec<&'a str>) {
    if depth > MAX_DEPTH {
        return;
    }
    match state {
        Value::String(text) => out.push(text),
        // Keys are ignored: they are usually English even when the values are not.
        Value::Object(map) => {
            for value in map.values() {
                iter_text(value, depth + 1, out);
            }
        }
        Value::Array(items) => {
            for value in items {
                iter_text(value, depth + 1, out);
            }
        }
        _ => {}
    }
}

/// Flatten a state into the text used for detection.
pub fn state_text(state: &Value) -> String {
    let mut leaves = Vec::new();
    iter_text(state, 0, &mut leaves);
    let joined = leaves.join(" ");
    // CPython slices by code point, not by byte.
    joined.chars().take(MAX_DETECTION_CHARS).collect()
}

/// Is this codepoint in the Latin range upstream treats as readable?
#[inline]
fn is_latin_codepoint(cp: u32) -> bool {
    cp < 0x0250 || (0x1E00..=0x1EFF).contains(&cp)
}

/// Per-script counts in first-seen order, plus the Latin count.
///
/// `latin_first` controls where the Latin entry sits, which is what decides
/// tie-breaking in `detect_script` and key order in `script_profile`.
fn count_scripts(text: &str, latin_first: bool) -> Vec<(String, usize)> {
    let mut counts: Vec<(String, usize)> = if latin_first {
        vec![("latin".to_string(), 0)]
    } else {
        Vec::new()
    };
    let mut latin = 0usize;

    for c in text.chars() {
        if !py_isalpha(c) {
            continue;
        }
        let cp = c as u32;
        if is_latin_codepoint(cp) {
            if latin_first {
                counts[0].1 += 1;
            } else {
                latin += 1;
            }
            continue;
        }
        for (name, ranges) in SCRIPT_RANGES {
            if ranges.iter().any(|(lo, hi)| (*lo..=*hi).contains(&cp)) {
                match counts.iter_mut().find(|(key, _)| key == name) {
                    Some(entry) => entry.1 += 1,
                    None => counts.push((name.to_string(), 1)),
                }
                break;
            }
        }
    }

    if !latin_first {
        // Upstream assigns `counts["latin"]` after the loop, so it lands last
        // and loses ties against any script already present.
        match counts.iter_mut().find(|(key, _)| key == "latin") {
            Some(entry) => entry.1 = latin,
            None => counts.push(("latin".to_string(), latin)),
        }
    }
    counts
}

/// Dominant script: `"latin"`, `"han"`, … or `"unknown"` when there are no letters.
pub fn detect_script(text: &str) -> String {
    let counts = count_scripts(text, false);
    let total: usize = counts.iter().map(|(_, n)| *n).sum();
    if total == 0 {
        return "unknown".to_string();
    }
    // `max()` keeps the first maximum in iteration order.
    counts
        .into_iter()
        .max_by_key(|(_, n)| *n)
        .map(|(name, _)| name)
        .unwrap_or_else(|| "unknown".to_string())
}

/// Fraction of alphabetic characters belonging to each detected script.
pub fn script_profile(text: &str) -> Vec<(String, f64)> {
    let counts = count_scripts(text, true);
    let total: usize = counts.iter().map(|(_, n)| *n).sum();
    if total == 0 {
        return Vec::new();
    }
    counts
        .into_iter()
        .filter(|(_, n)| *n > 0)
        .map(|(name, n)| (name, n as f64 / total as f64))
        .collect()
}

/// Best-effort language code for Latin-script text, or `None` when undecided.
pub fn guess_latin_language(text: &str) -> Option<String> {
    let words = py_words_lowercase(text);
    if words.len() < 4 {
        return None;
    }

    let scores: Vec<(&str, usize)> = STOPWORDS
        .iter()
        .map(|(code, stopwords)| {
            let hits = words
                .iter()
                .filter(|word| stopwords.binary_search(&word.as_str()).is_ok())
                .count();
            (*code, hits)
        })
        .collect();

    let lowered = text.to_lowercase();
    let diacritics = lowered
        .chars()
        .filter(|c| NON_EN_DIACRITICS.contains(*c))
        .count();
    let diacritic_rate = diacritics as f64 / std::cmp::max(1, lowered.chars().count()) as f64;

    let english = scores
        .iter()
        .find(|(code, _)| *code == "en")
        .map_or(0, |(_, hits)| *hits);

    // `max()` over the non-English entries, first maximum wins.
    let best = scores
        .iter()
        .filter(|(code, _)| *code != "en")
        .copied()
        .fold(None::<(&str, usize)>, |acc, item| match acc {
            Some((_, best)) if best >= item.1 => acc,
            _ => Some(item),
        });
    let (best_lang, best_score) = match best {
        Some((lang, score)) => (Some(lang), score),
        None => (None, 0),
    };

    let fallback = || {
        if english > 0 {
            Some("en".to_string())
        } else {
            None
        }
    };

    if best_score == 0 && diacritic_rate < 0.02 {
        return fallback();
    }
    if let Some(lang) = best_lang {
        // A non-English language needs a clear margin over English function words.
        if best_score >= std::cmp::max(2, english + 2) {
            return Some(lang.to_string());
        }
        if diacritic_rate >= 0.04 && best_score >= english {
            return Some(lang.to_string());
        }
    }
    fallback()
}

/// Full detection result for a state.
pub fn analyse(state: &Value) -> Detection {
    let text = state_text(state);
    let profile = script_profile(&text);
    let script = detect_script(&text);

    let mut detection = Detection {
        script: script.clone(),
        script_profile: profile,
        language: None,
        is_english: true,
        non_latin_fraction: 0.0,
    };

    if script == "unknown" {
        return detection;
    }

    let non_latin = if detection.script_profile.is_empty() {
        0.0
    } else {
        py_round(1.0 - detection.latin_fraction(), 4)
    };

    if script != "latin" {
        detection.is_english = false;
        detection.non_latin_fraction = non_latin;
        return detection;
    }

    let language = guess_latin_language(&text);
    detection.is_english = matches!(language.as_deref(), None | Some("en"));
    detection.language = language;
    detection.non_latin_fraction = non_latin;
    detection
}

/// True when the English checkpoint can be expected to read this state.
pub fn is_english(state: &Value) -> bool {
    analyse(state).is_english
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn detects_scripts() {
        assert_eq!(detect_script("hello there"), "latin");
        assert_eq!(detect_script("Мой счёт"), "cyrillic");
        assert_eq!(detect_script("こんにちは"), "kana");
        assert_eq!(detect_script("मेरा खाता"), "devanagari");
        assert_eq!(detect_script("12345 !!!"), "unknown");
    }

    #[test]
    fn combining_marks_do_not_count_as_letters() {
        // U+093E is a combining vowel sign: CPython's isalpha() rejects it, so
        // it must not inflate the Devanagari share.
        let profile = script_profile("\u{0915}\u{093E}");
        assert_eq!(profile, vec![("devanagari".to_string(), 1.0)]);
    }

    #[test]
    fn english_text_routes_to_english() {
        let detection = analyse(&json!("I was charged twice and would like a refund please"));
        assert_eq!(detection.script, "latin");
        assert!(detection.is_english);
    }

    #[test]
    fn german_text_is_not_english() {
        let detection = analyse(&json!(
            "Mein Konto wurde zweimal belastet und ich möchte eine Rückerstattung"
        ));
        assert_eq!(detection.script, "latin");
        assert!(!detection.is_english);
        assert_eq!(detection.language.as_deref(), Some("de"));
    }

    #[test]
    fn non_latin_reports_a_fraction() {
        let detection = analyse(&json!("Мой счёт был дважды списан"));
        assert!(!detection.is_english);
        assert!(detection.non_latin_fraction > 0.9);
    }

    #[test]
    fn nested_states_are_flattened() {
        let detection = analyse(&json!({"message": {"body": ["こんにちは", 42]}}));
        assert_eq!(detection.script, "kana");
    }

    #[test]
    fn short_input_stays_undecided() {
        assert_eq!(guess_latin_language("hi there"), None);
    }
}