Function decode_source_bytes

crates/vm/src/stdlib/builtins.rs:192–249

(source: &[u8], filename: &str, vm: &VirtualMachine)

    #[cfg(feature = "parser")]
    fn decode_source_bytes(source: &[u8], filename: &str, vm: &VirtualMachine) -> PyResult<String> {
        let has_bom = source.starts_with(b"\xef\xbb\xbf");
        let encoding = detect_source_encoding(source);

        let is_utf8 = encoding.as_deref().is_none_or(is_utf8_encoding);

        // Validate BOM + encoding combination
        if has_bom && !is_utf8 {
            return Err(vm.new_exception_msg(
                vm.ctx.exceptions.syntax_error.to_owned(),
                format!("encoding problem for '{filename}': utf-8").into(),
            ));
        }

        if is_utf8 {
            let src = if has_bom { &source[3..] } else { source };
            match core::str::from_utf8(src) {
                Ok(s) => Ok(s.to_owned()),
                Err(e) => {
                    let bad_byte = src[e.valid_up_to()];
                    let line = src[..e.valid_up_to()]
                        .iter()
                        .filter(|&&b| b == b'\n')
                        .count()
                        + 1;
                    Err(vm.new_exception_msg(
                        vm.ctx.exceptions.syntax_error.to_owned(),
                        format!(
                            "Non-UTF-8 code starting with '\\x{bad_byte:02x}' \
                             on line {line}, but no encoding declared; \
                             see https://peps.python.org/pep-0263/ for details \
                             ({filename}, line {line})"
                        )
                        .into(),
                    ))
                }
            }
        } else {
            // Use codec registry for non-UTF-8 encodings
            let enc = encoding.as_deref().unwrap();
            let bytes_obj = vm.ctx.new_bytes(source.to_vec());
            let decoded = vm
                .state
                .codec_registry
                .decode_text(bytes_obj.into(), enc, None, vm)
                .map_err(|exc| {
                    if exc.fast_isinstance(vm.ctx.exceptions.lookup_error) {
                        vm.new_exception_msg(
                            vm.ctx.exceptions.syntax_error.to_owned(),
                            format!("unknown encoding for '{filename}': {enc}").into(),
                        )
                    } else {
                        exc
                    }
                })?;
            Ok(decoded.to_string_lossy().into_owned())
        }
    }