08_clean_source_code.py

from pathlib import Path
import re
import sys

import pandas as pd

PROJECT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(PROJECT_DIR))

import importlib

audit = importlib.import_module("07_audit_source_code")

INPUT_PATH = PROJECT_DIR / "SBAN_clean.parquet"
OUTPUT_PATH = PROJECT_DIR / "SBAN_clean.parquet"
REPORT_DIR = PROJECT_DIR / "reports" / "audit"

ID_COL = audit.ID_COL
LABEL_COL = audit.LABEL_COL
COLUMN = audit.COLUMN

META_COMMENT_HINTS = (
    "no changes needed",
    "no changes are needed",
    "clean and modernized",
    "code to replace the provided",
    "provided input code",
)


def load_data() -> pd.DataFrame:
    if not INPUT_PATH.exists():
        raise FileNotFoundError(f"Dosya bulunamadı: {INPUT_PATH}")
    return pd.read_parquet(INPUT_PATH)


def collect_reasons(text: str) -> set[str]:
    reasons: list[str] = []
    reasons.extend(audit.check_prompt_phrases(text))
    reasons.extend(audit.check_assistant_phrases(text))
    reasons.extend(audit.check_comment_instructions(text))
    reasons.extend(audit.check_structure(text))
    return set(reasons)


def source_stats(text: str) -> dict[str, int]:
    non_comment = audit.non_comment_lines(text)
    placeholder_comment = 0
    for line in str(text).splitlines():
        stripped = line.strip()
        if not stripped.startswith("//"):
            continue
        if audit.PLACEHOLDER_RE.search(stripped):
            placeholder_comment += 1
        elif any(h in stripped.lower() for h in audit.COMMENT_INSTRUCTION_HINTS):
            placeholder_comment += 1
    return {
        "n_non_comment": len(non_comment),
        "non_comment_chars": sum(len(line) for line in non_comment),
        "placeholder_comment": placeholder_comment,
    }


def is_false_positive(text: str, reasons: set[str]) -> bool:
    lower = str(text).lower()
    if "assistant_phrase:let me" in reasons and "printf" in lower:
        return True
    if "assistant_phrase:as follows" in reasons and len(audit.non_comment_lines(text)) >= 5:
        return True
    return False


def should_delete(text: str, reasons: set[str]) -> bool:
    if is_false_positive(text, reasons):
        return False

    lower = str(text).lower()
    if '#error "no code to output"' in lower or "#error 'no code to output'" in lower:
        return True

    if "comment_only" in reasons or not audit.non_comment_lines(text):
        return True

    stats = source_stats(text)
    if stats["n_non_comment"] <= 2 and stats["non_comment_chars"] < 80:
        return True
    if stats["placeholder_comment"] >= 3 and stats["non_comment_chars"] < 200:
        return True

    return False


def should_clean(text: str, reasons: set[str]) -> bool:
    if not reasons or is_false_positive(text, reasons):
        return False
    if should_delete(text, reasons):
        return False
    return bool(audit.non_comment_lines(text))


def should_drop_comment_line(stripped: str) -> bool:
    if not stripped.startswith("//"):
        return False

    lower = stripped.lower()
    if any(h in lower for h in audit.COMMENT_INSTRUCTION_HINTS):
        return True
    if any(h in lower for h in META_COMMENT_HINTS):
        return True
    if audit.PLACEHOLDER_RE.search(stripped):
        return True
    if "<complete_cleaned_code>" in lower:
        return True
    if any(phrase in lower for phrase in audit.ASSISTANT_PHRASES):
        return True
    return False


def clean_source_text(text: str) -> str:
    cleaned_lines = []
    for line in str(text).splitlines():
        stripped = line.strip()
        if stripped and should_drop_comment_line(stripped):
            continue
        cleaned_lines.append(line)

    cleaned = "\n".join(cleaned_lines)
    cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
    return cleaned.strip()


def classify_action(text: str, reasons: set[str] | None = None) -> str:
    if reasons is None:
        reasons = collect_reasons(text)
    if not reasons:
        return "unchanged"
    if should_delete(text, reasons):
        return "delete"
    if should_clean(text, reasons):
        return "clean"
    return "keep"


def apply_cleaning(dataframe: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    audit_df = audit.audit_source(dataframe[[ID_COL, LABEL_COL, COLUMN]])
    flagged_ids = set(audit_df[ID_COL])

    action_rows = []
    delete_ids: set[str] = set()
    clean_map: dict[str, str] = {}

    for _, row in dataframe.iterrows():
        sample_id = row[ID_COL]
        if sample_id not in flagged_ids:
            action_rows.append(
                {ID_COL: sample_id, LABEL_COL: row[LABEL_COL], "action": "unchanged"}
            )
            continue

        text = row[COLUMN]
        reasons = collect_reasons(text)
        action = classify_action(text, reasons)
        action_rows.append({ID_COL: sample_id, LABEL_COL: row[LABEL_COL], "action": action})

        if action == "delete":
            delete_ids.add(sample_id)
        elif action == "clean":
            clean_map[sample_id] = clean_source_text(text)

    cleaned_records = []
    for index, row in dataframe.iterrows():
        sample_id = row[ID_COL]
        if sample_id not in clean_map or sample_id in delete_ids:
            continue

        original_text = row[COLUMN]
        cleaned_text = clean_map[sample_id]
        post_reasons = collect_reasons(cleaned_text)
        if should_delete(cleaned_text, post_reasons):
            delete_ids.add(sample_id)
            continue

        dataframe.at[index, COLUMN] = cleaned_text
        cleaned_records.append(
            {
                ID_COL: sample_id,
                LABEL_COL: row[LABEL_COL],
                "before_len": len(str(original_text)),
                "after_len": len(cleaned_text),
            }
        )

    action_df = pd.DataFrame(action_rows)
    action_df.loc[action_df[ID_COL].isin(delete_ids), "action"] = "delete"
    cleaned_dataframe = dataframe[~dataframe[ID_COL].isin(delete_ids)].reset_index(drop=True)
    return cleaned_dataframe, action_df, pd.DataFrame(cleaned_records)


def print_summary(before: int, after: int, action_df: pd.DataFrame, cleaned_df: pd.DataFrame) -> None:
    print(f"Önce:  {before:,} satır")
    print(f"Sonra: {after:,} satır")
    print(f"Silinen: {before - after:,}")
    print(f"Temizlenen source: {len(cleaned_df):,}")
    print("\nAksiyon dağılımı:")
    print(action_df["action"].value_counts().to_string())


def main() -> None:
    dataframe = load_data()
    before_count = len(dataframe)

    cleaned_dataframe, action_df, cleaned_details = apply_cleaning(dataframe)

    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    action_df.to_csv(REPORT_DIR / "source_clean_actions.csv", index=False)
    cleaned_details.to_csv(REPORT_DIR / "source_cleaned_ids.csv", index=False)
    action_df.loc[action_df["action"] == "delete", [ID_COL, LABEL_COL]].to_csv(
        REPORT_DIR / "source_deleted_ids.csv",
        index=False,
    )

    cleaned_dataframe.to_parquet(OUTPUT_PATH, index=False)

    print(f"Girdi:  {INPUT_PATH.name}")
    print_summary(before_count, len(cleaned_dataframe), action_df, cleaned_details)
    print(f"\nKaydedildi: {OUTPUT_PATH}")
    print(f"Raporlar: {REPORT_DIR}")


if __name__ == "__main__":
    main()