Farhannr28/code-authorship-demo-space

1from __future__ import annotations2 3import importlib4import os5import sys6from dataclasses import dataclass7from functools import lru_cache8from pathlib import Path9from typing import Any10 11LANGUAGES = ("Python", "Java", "C++", "Go", "PHP", "C#", "C", "JavaScript")12LANGUAGE_ALIASES = {13    **{language.casefold(): language for language in LANGUAGES},14    "cpp": "C++",15    "csharp": "C#",16    "c_sharp": "C#",17    "golang": "Go",18    "js": "JavaScript",19    "py": "Python",20}21 22 23@dataclass(frozen=True)24class PreprocessedInput:25    language: str26    code_view: str27    language_view: str28    input_characters: int29    input_lines: int30    trace: tuple[dict[str, Any], ...]31 32    def api_response(self) -> dict[str, Any]:33        return {34            "schema_version": "1.0",35            "language": self.language,36            "input": {37                "characters": self.input_characters,38                "lines": self.input_lines,39            },40            "features": {41                "code_view": self.code_view,42                "language_view": self.language_view,43            },44            "trace": list(self.trace),45        }46 47 48def normalize_language(language: str) -> str:49    if not isinstance(language, str):50        raise TypeError("Language must be a string.")51    resolved = LANGUAGE_ALIASES.get(language.strip().casefold())52    if resolved is None:53        raise ValueError(f"Unsupported language {language!r}. Choose from {LANGUAGES}.")54    return resolved55 56 57def preprocess_source(code: str, language: str) -> PreprocessedInput:58    if not isinstance(code, str):59        raise TypeError("Source code must be text.")60    if not code.strip():61        raise ValueError("Source code cannot be empty or whitespace-only.")62    language = normalize_language(language)63    inference_features, pipeline_utils = _implementation_modules()64 65    try:66        values = inference_features.extract_formatting_features(code, language)67    except LookupError as exc:68        raise RuntimeError(69            "Required NLTK resources are unavailable. Install punkt_tab and "70            "averaged_perceptron_tagger_eng."71        ) from exc72 73    (74        trailing,75        indent,76        ending,77        blank,78        mixed,79        density,80        leading,81        verbs,82        leak,83        preamble,84        postamble,85        has_preamble,86        has_postamble,87    ) = values88    categories = inference_features.categorize_stylometric_ratios(89        trailing, blank, density, leading, verbs, leak90    )91    category_names = (92        "trailing_ws_ratio_cat",93        "blank_line_ratio_cat",94        "comment_pos_cat",95        "comment_density_cat",96        "comment_verb_cat",97        "lang_leak_cat",98    )99    record = dict(zip(category_names, categories))100    clean_body = inference_features._split_preamble_postamble(code, language)[0]101    record.update(102        code_for_parsing=clean_body,103        language=language,104        indent_type=indent,105        line_ending=ending,106        mixed_indentation=mixed,107        preamble_text=preamble,108        postamble_text=postamble,109        has_preamble=has_preamble,110        has_postamble=has_postamble,111    )112    views = pipeline_utils.process_nl_batch(113        {key: [value] for key, value in record.items()}114    )115 116    formatting = {117        "indent_type": indent,118        "line_ending": ending,119        "mixed_indentation": mixed,120        **dict(zip(category_names, categories)),121    }122    trace = (123        {124            "stage": "preamble_postamble",125            "preamble_detected": has_preamble,126            "postamble_detected": has_postamble,127        },128        {"stage": "formatting_features", **formatting},129        {130            "stage": "view_assembly",131            "code_view_characters": len(views["sanitized"][0]),132            "language_view_characters": len(views["natural_language"][0]),133        },134    )135    return PreprocessedInput(136        language=language,137        code_view=views["sanitized"][0],138        language_view=views["natural_language"][0],139        input_characters=len(code),140        input_lines=code.count("\n") + 1,141        trace=trace,142    )143 144 145def ensure_nltk_resources() -> None:146    import nltk147 148    data_dir = Path(os.environ.get("NLTK_DATA", "/tmp/nltk_data"))149    data_dir.mkdir(parents=True, exist_ok=True)150    if str(data_dir) not in nltk.data.path:151        nltk.data.path.insert(0, str(data_dir))152    resources = (153        ("tokenizers/punkt_tab/english", "punkt_tab"),154        ("taggers/averaged_perceptron_tagger_eng", "averaged_perceptron_tagger_eng"),155    )156    for resource, package in resources:157        try:158            nltk.data.find(resource)159        except LookupError:160            if not nltk.download(package, download_dir=str(data_dir), quiet=True):161                raise RuntimeError(162                    f"Could not download required NLTK resource {package!r}."163                )164 165 166@lru_cache(maxsize=1)167def _implementation_modules():168    implementation = Path(__file__).resolve().parent / "preprocessing_impl"169    location = str(implementation)170    if location not in sys.path:171        sys.path.insert(0, location)172    return (173        importlib.import_module("inference_features"),174        importlib.import_module("pipeline_utils"),175    )176