07_audit_source_code.py

from pathlib import Path
import re

import pandas as pd

PROJECT_DIR = Path(__file__).resolve().parent
INPUT_PATH = PROJECT_DIR / "SBAN_clean.parquet"
REPORT_DIR = PROJECT_DIR / "reports" / "audit"

COLUMN = "source_code"
ID_COL = "ID"
LABEL_COL = "dataset_name"

PROMPT_PHRASES = (
    "paste the full source code",
    "insert source code",
    "paste code",
    "your code",
    "add main function",
    "code goes",
    "implementation goes",
    "insert code",
    "no modifications needed",
    "corrected",
    "here",
    "no comments",
    "additional",
)

ASSISTANT_PHRASES = (
    "here is the",
    "here's the",
    "below is",
    "as follows",
    "let me",
    "sure!",
    "certainly",
    "without comments",
    "only the code",
    "only code",
    "corrected version",
    "fix the code",
    "complete_cleaned_code",
    "main function is missing",
    "no c++ solution",
    "no output as the code",
)

COMMENT_INSTRUCTION_HINTS = (
    "please remove",
    "finish the code",
    "add necessary",
    "replace windows",
    "missing headers",
    "final code",
    "resolve syntax",
    "assume ",
    "provided below",
    "no c++ solution",
    "complete_cleaned",
)

CODE_INDICATOR_RE = re.compile(
    r"#include|#define|\{|;|\b(void|int|char|return|main|struct|class|typedef|enum|if|for|while)\b|__cdecl|__fastcall",
    re.IGNORECASE,
)

PLACEHOLDER_RE = re.compile(
    r"//\s*\.{3}|//\s*\.\.\.\s*$|<complete_cleaned_code>|//\s*\.\.\.\s*\n",
    re.IGNORECASE,
)


def load_data() -> pd.DataFrame:
    if not INPUT_PATH.exists():
        raise FileNotFoundError(f"Dosya bulunamadı: {INPUT_PATH}")
    return pd.read_parquet(INPUT_PATH, columns=[ID_COL, LABEL_COL, COLUMN])


def non_comment_lines(text: str) -> list[str]:
    lines = []
    for line in str(text).splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        if stripped.startswith("//"):
            continue
        if stripped.startswith("/*") or stripped.startswith("*"):
            continue
        lines.append(stripped)
    return lines


def has_code_indicators(text: str) -> bool:
    return bool(CODE_INDICATOR_RE.search(str(text)))


def check_prompt_phrases(text: str) -> list[str]:
    reasons = []
    lower = str(text).lower()
    for phrase in PROMPT_PHRASES:
        if phrase in lower:
            reasons.append(f"prompt_phrase:{phrase}")
    return reasons


def check_assistant_phrases(text: str) -> list[str]:
    reasons = []
    lower = str(text).lower()
    for phrase in ASSISTANT_PHRASES:
        if phrase in lower:
            reasons.append(f"assistant_phrase:{phrase}")
    return reasons


def check_comment_instructions(text: str) -> list[str]:
    reasons = []
    for line in str(text).splitlines():
        stripped = line.strip()
        if not stripped.startswith("//"):
            continue
        lower = stripped.lower()
        for hint in COMMENT_INSTRUCTION_HINTS:
            if hint in lower:
                reasons.append(f"instruction_comment:{hint}")
                break
    return reasons


def check_structure(text: str) -> list[str]:
    text = str(text)
    reasons = []

    if "```" in text:
        reasons.append("markdown_fence")

    if PLACEHOLDER_RE.search(text):
        reasons.append("placeholder_marker")

    if len(text.strip()) < 20:
        reasons.append("too_short:lt20")

    if not non_comment_lines(text):
        reasons.append("comment_only")

    if not has_code_indicators(text):
        reasons.append("no_code_indicators")

    if len(text.strip()) < 50 and not has_code_indicators(text):
        reasons.append("short_without_code")

    return reasons


def audit_source(dataframe: pd.DataFrame) -> pd.DataFrame:
    records = []

    for _, row in dataframe.iterrows():
        text = row[COLUMN]
        reasons = []
        reasons.extend(check_prompt_phrases(text))
        reasons.extend(check_assistant_phrases(text))
        reasons.extend(check_comment_instructions(text))
        reasons.extend(check_structure(text))

        if not reasons:
            continue

        unique_reasons = sorted(set(reasons))
        high_reasons = {
            "comment_only",
            "no_code_indicators",
            "markdown_fence",
            "too_short:lt20",
            "short_without_code",
        }
        severity = "high" if any(
            reason.split(":")[0] in high_reasons
            or reason.startswith("prompt_phrase:")
            for reason in unique_reasons
        ) else "medium"

        preview = str(text).replace("\n", " ")[:200]
        records.append(
            {
                ID_COL: row[ID_COL],
                LABEL_COL: row[LABEL_COL],
                "severity": severity,
                "reasons": "|".join(unique_reasons),
                "reason_count": len(unique_reasons),
                "preview": preview,
            }
        )

    return pd.DataFrame(records)


def print_summary(flagged: pd.DataFrame, total_rows: int) -> None:
    print(f"Toplam satır: {total_rows:,}")
    print(f"Flagged ID: {len(flagged):,} ({len(flagged) / total_rows:.2%})")
    print(f"  high: {(flagged['severity'] == 'high').sum():,}")
    print(f"  medium: {(flagged['severity'] == 'medium').sum():,}")

    print("\nDataset kırılımı:")
    print(flagged[LABEL_COL].value_counts().sort_index().to_string())

    reason_counts: dict[str, int] = {}
    for reasons in flagged["reasons"]:
        for reason in reasons.split("|"):
            reason_counts[reason] = reason_counts.get(reason, 0) + 1

    print("\nNeden sayıları:")
    for reason, count in sorted(reason_counts.items(), key=lambda item: -item[1]):
        print(f"  {reason}: {count:,}")


def main() -> None:
    dataframe = load_data()
    flagged = audit_source(dataframe)

    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    output_path = REPORT_DIR / "source_flagged_ids.csv"
    flagged.sort_values([ "severity", LABEL_COL, ID_COL], ascending=[True, True, True]).to_csv(
        output_path, index=False
    )

    ids_path = REPORT_DIR / "source_flagged_ids.txt"
    ids_path.write_text("\n".join(flagged[ID_COL].astype(str).tolist()) + "\n")

    print(f"Girdi: {INPUT_PATH.name}")
    print_summary(flagged, len(dataframe))
    print(f"\nKaydedildi:\n  {output_path}\n  {ids_path}")


if __name__ == "__main__":
    main()