Log Struct Splitter.py

# Log Struct Splitter.py
# Detect recurring header lines and split logs into blocks automatically.
# Input: .csv/.txt files in the same directory as this script (or exe).
# Output: output_split/ (split CSVs + report)

from __future__ import annotations

import re
import csv
import sys
from pathlib import Path
from dataclasses import dataclass
from collections import Counter, defaultdict
from typing import List, Tuple, Optional, Dict, Any


# -----------------------------
# Config
# -----------------------------
SANITIZE_OUTPUT_FILENAMES = True          # safer for GitHub/Windows tooling
CONFIDENCE_WARN_THRESHOLD = 60            # show WARNING if below this
MAX_STEM_LEN = 80


# -----------------------------
# Utilities
# -----------------------------
def get_base_dir() -> Path:
    """Return the directory where the script/exe is located."""
    if getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS"):
        # PyInstaller onefile: sys.executable points to the exe location
        return Path(sys.executable).resolve().parent
    return Path(__file__).resolve().parent


def read_text_lines(path: Path) -> List[str]:
    """Robust text read with fallback encodings."""
    for enc in ("utf-8-sig", "utf-8", "cp932", "shift_jis", "latin-1"):
        try:
            return path.read_text(encoding=enc, errors="strict").splitlines()
        except Exception:
            continue
    # last resort
    return path.read_text(encoding="utf-8", errors="replace").splitlines()


def is_comment_or_blank(line: str) -> bool:
    s = line.strip()
    if not s:
        return True
    return s.startswith(("#", "//", ";"))


def normalize_header(line: str) -> str:
    # normalize spaces, case
    s = line.strip().lower()
    s = re.sub(r"\s+", "", s)
    return s


def try_parse_float(token: str) -> bool:
    t = token.strip()
    if not t:
        return False
    try:
        float(t)
        return True
    except Exception:
        return False


def _variance(nums: List[int]) -> float:
    if not nums:
        return 0.0
    m = sum(nums) / len(nums)
    return sum((x - m) ** 2 for x in nums) / len(nums)


def tokenize(line: str, delim: str) -> List[str]:
    return [t.strip() for t in line.strip().split(delim)]


def guess_delimiter(lines: List[str]) -> str:
    """
    Guess delimiter by scoring candidate delimiters on non-comment lines.
    Prefers the delimiter that yields consistent column counts.
    """
    candidates = [",", "\t", ";", "|"]
    sample = [ln for ln in lines if not is_comment_or_blank(ln)]
    sample = sample[:2000]  # cap
    if not sample:
        return ","

    best_delim = ","
    best_score = -1.0

    for d in candidates:
        col_counts = []
        for ln in sample[:200]:
            parts = ln.split(d)
            if len(parts) >= 2:
                col_counts.append(len(parts))
        if len(col_counts) < 10:
            continue

        mode = Counter(col_counts).most_common(1)[0][0]
        dominance = col_counts.count(mode) / len(col_counts)
        variance = _variance(col_counts)
        score = dominance * 10.0 - variance

        if score > best_score:
            best_score = score
            best_delim = d

    return best_delim


def safe_stem(name: str, max_len: int = MAX_STEM_LEN) -> str:
    """
    Sanitize output filename stem to avoid problematic characters for tools/OS.
    """
    s = name.strip()
    # Replace Windows-illegal characters
    s = re.sub(r'[\\/:*?"<>|]', "_", s)
    # Also normalize common troublemakers
    s = s.replace(",", "_").replace("#", "").replace("\t", "_")
    s = re.sub(r"\s+", "_", s).strip("_")
    if not s:
        s = "log"
    return s[:max_len]


# -----------------------------
# Header detection logic
# -----------------------------
@dataclass
class HeaderCandidate:
    idx: int
    raw: str
    norm: str
    n_cols: int
    alpha_ratio: float  # proportion of non-numeric tokens


def _alpha_ratio(tokens: List[str]) -> float:
    non_numeric = 0
    non_empty = 0
    for t in tokens:
        if t == "":
            continue
        non_empty += 1
        if not try_parse_float(t):
            non_numeric += 1
    if non_empty == 0:
        return 0.0
    return non_numeric / non_empty


def header_candidate_score(tokens: List[str]) -> float:
    """
    Score likelihood that a line is a header.
    Higher is more likely.
    """
    if len(tokens) < 2:
        return -1e9

    non_numeric = 0
    non_empty = 0
    for t in tokens:
        if t == "":
            continue
        non_empty += 1
        if not try_parse_float(t):
            non_numeric += 1

    if non_empty == 0:
        return -1e9

    alpha_ratio = non_numeric / non_empty

    # Penalize if many tokens look like timestamps/dates (still could be header)
    ts_like = sum(bool(re.search(r"\d{4}[-/]\d{1,2}[-/]\d{1,2}", t)) for t in tokens)
    time_like = sum(bool(re.search(r"\d{1,2}:\d{2}(:\d{2})?", t)) for t in tokens)

    score = 0.0
    score += alpha_ratio * 5.0
    score += min(len(tokens), 50) * 0.05
    score -= (ts_like + time_like) * 0.2

    joined = " ".join(tokens).lower()
    if any(k in joined for k in ["time", "date", "temp", "pressure", "id", "voltage", "current", "power", "step"]):
        score += 0.5

    return score


def detect_recurring_header(lines: List[str], delim: str) -> Tuple[Optional[str], List[int], Dict[str, Any]]:
    """
    Detect the most plausible recurring header by:
    1) collecting header candidates
    2) picking candidates that recur
    3) choosing the best by (frequency, header score, column consistency)
    Returns: (selected_header_norm, positions, stats)
    """
    candidates: List[HeaderCandidate] = []
    norm_counts = Counter()
    norm_best_score = defaultdict(lambda: -1e9)
    norm_cols: Dict[str, int] = {}

    for i, ln in enumerate(lines):
        if is_comment_or_blank(ln):
            continue
        toks = tokenize(ln, delim)
        sc = header_candidate_score(toks)
        if sc < 2.0:  # threshold
            continue
        norm = normalize_header(ln)
        ar = _alpha_ratio(toks)
        candidates.append(HeaderCandidate(i, ln.rstrip("\n"), norm, len(toks), ar))
        norm_counts[norm] += 1
        if sc > norm_best_score[norm]:
            norm_best_score[norm] = sc
            norm_cols[norm] = len(toks)

    recurring = [n for n, c in norm_counts.items() if c >= 2]
    if not recurring:
        return None, [], {"reason": "no recurring header candidates found", "candidates": len(candidates)}

    def key_fn(n: str):
        return (
            norm_counts[n],
            norm_best_score[n],
            norm_cols.get(n, 0),
        )

    selected = max(recurring, key=key_fn)
    positions = [c.idx for c in candidates if c.norm == selected]

    stats: Dict[str, Any] = {
        "selected_freq": int(norm_counts[selected]),
        "unique_candidate_headers": int(len(norm_counts)),
        "total_candidate_lines": int(len(candidates)),
        "selected_cols": int(norm_cols.get(selected, 0)),
    }
    return selected, positions, stats


# -----------------------------
# Splitting
# -----------------------------
def split_by_recurring_header(lines: List[str], delim: str, selected_header_norm: str) -> List[List[str]]:
    blocks: List[List[str]] = []
    current: List[str] = []
    header_seen = False

    for ln in lines:
        raw = ln.rstrip("\r\n")

        if not is_comment_or_blank(raw):
            if normalize_header(raw) == selected_header_norm:
                if header_seen and current:
                    blocks.append(current)
                    current = []
                header_seen = True

        current.append(raw)

    if current:
        blocks.append(current)

    # Trim preamble before first header
    if blocks:
        first = blocks[0]
        cut = 0
        for i, ln in enumerate(first):
            if not is_comment_or_blank(ln) and normalize_header(ln) == selected_header_norm:
                cut = i
                break
        blocks[0] = first[cut:]

    # Keep only blocks that contain the header (safety)
    cleaned = []
    for b in blocks:
        if any((not is_comment_or_blank(ln)) and normalize_header(ln) == selected_header_norm for ln in b):
            cleaned.append(b)
    return cleaned


def block_column_stats(block: List[str], delim: str) -> Dict[str, Any]:
    """
    Compute per-block column consistency stats.
    mismatch_rate: fraction of non-comment lines whose column count != mode.
    """
    col_counts = []
    non_comment_lines = 0
    for ln in block:
        if is_comment_or_blank(ln):
            continue
        non_comment_lines += 1
        col_counts.append(len(tokenize(ln, delim)))

    if not col_counts:
        return {"rows": 0, "cols_mode": 0, "mismatch_rate": 0.0, "mode_count": 0}

    cnt = Counter(col_counts)
    cols_mode, mode_count = cnt.most_common(1)[0]
    mismatch = 1.0 - (mode_count / len(col_counts)) if len(col_counts) else 0.0
    return {
        "rows": int(non_comment_lines),
        "cols_mode": int(cols_mode),
        "mode_count": int(mode_count),
        "mismatch_rate": float(mismatch),
        "unique_col_counts": dict(sorted(cnt.items(), key=lambda x: x[0])),
    }


def write_block_csv(block: List[str], out_path: Path, delim: str) -> int:
    """
    Write a block into CSV (comma-delimited) while preserving tokens.
    Returns: n_rows_written (non-comment lines)
    """
    out_path.parent.mkdir(parents=True, exist_ok=True)
    n_rows = 0
    with out_path.open("w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        for ln in block:
            if is_comment_or_blank(ln):
                continue
            toks = tokenize(ln, delim)
            w.writerow(toks)
            n_rows += 1
    return n_rows


# -----------------------------
# Confidence scoring
# -----------------------------
def compute_confidence_score(
    total_lines: int,
    noise_lines: int,
    header_occurrences: int,
    blocks_count: int,
    per_block_stats: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """
    Return: {"score": int, "breakdown": {...}, "notes": [...]}
    Score is 0..100.
    """
    notes: List[str] = []

    # Noise ratio
    noise_ratio = (noise_lines / total_lines) if total_lines > 0 else 0.0

    # Column consistency (weighted by rows)
    total_rows = sum(s.get("rows", 0) for s in per_block_stats) or 0
    if total_rows > 0:
        weighted_consistency = 0.0
        for s in per_block_stats:
            rows = s.get("rows", 0)
            mismatch = float(s.get("mismatch_rate", 0.0))
            consistency = max(0.0, min(1.0, 1.0 - mismatch))
            weighted_consistency += consistency * rows
        col_consistency = weighted_consistency / total_rows
    else:
        col_consistency = 0.0

    # Header frequency score (max 30)
    if header_occurrences <= 1:
        header_score = 0.0
        notes.append("Header does not recur (<=1). Splitting confidence is low.")
    elif header_occurrences == 2:
        header_score = 20.0
    elif header_occurrences == 3:
        header_score = 27.0
    else:
        header_score = 30.0

    # Header/block alignment (max 20)
    # Ideal: blocks == header_occurrences
    diff = abs(blocks_count - header_occurrences)
    align_score = max(0.0, 20.0 - diff * 10.0)
    if diff > 0:
        notes.append(f"Blocks ({blocks_count}) and header occurrences ({header_occurrences}) differ by {diff}.")

    # Column consistency (max 40)
    col_score = 40.0 * col_consistency
    if col_consistency < 0.9:
        notes.append(f"Column consistency is {col_consistency:.2f}. Some rows may be malformed or missing fields.")

    # Noise penalty (max 10)
    # If noise_ratio <= 0.10 => near full
    # If noise_ratio >= 0.50 => 0
    if noise_ratio <= 0.10:
        noise_score = 10.0
    elif noise_ratio >= 0.50:
        noise_score = 0.0
        notes.append(f"High noise ratio: {noise_ratio:.2f} (comments/blank lines).")
    else:
        noise_score = 10.0 * (1.0 - (noise_ratio - 0.10) / (0.40))

    total = header_score + align_score + col_score + noise_score
    score = int(round(max(0.0, min(100.0, total))))

    breakdown = {
        "header_freq": round(header_score, 1),
        "header_block_alignment": round(align_score, 1),
        "column_consistency": round(col_score, 1),
        "noise_ratio": round(noise_score, 1),
        "column_consistency_ratio": round(col_consistency, 3),
        "noise_ratio_raw": round(noise_ratio, 3),
    }

    return {"score": score, "breakdown": breakdown, "notes": notes}


# -----------------------------
# Main
# -----------------------------
def process_file(path: Path, out_dir: Path) -> str:
    lines = read_text_lines(path)
    delim = guess_delimiter(lines)

    total_lines = len(lines)
    noise_lines = sum(1 for ln in lines if is_comment_or_blank(ln))

    selected_norm, positions, stats = detect_recurring_header(lines, delim)
    report_lines: List[str] = []
    report_lines.append(f"FILE: {path.name}")
    report_lines.append(f"  delimiter_guess: {repr(delim)}")

    if selected_norm is None:
        report_lines.append("  result: NO_SPLIT (no recurring header detected)")
        report_lines.append(f"  details: {stats}")
        # Even for NO_SPLIT, provide a minimal confidence hint
        conf = compute_confidence_score(
            total_lines=total_lines,
            noise_lines=noise_lines,
            header_occurrences=0,
            blocks_count=0,
            per_block_stats=[],
        )
        report_lines.append(f"  confidence_score: {conf['score']}/100")
        report_lines.append(f"  confidence_breakdown: {conf['breakdown']}")
        if conf["notes"]:
            report_lines.append(f"  confidence_notes: {conf['notes']}")
        return "\n".join(report_lines)

    blocks = split_by_recurring_header(lines, delim, selected_norm)

    report_lines.append("  result: SPLIT")
    report_lines.append(
        f"  header_occurrences: {len(positions)} at lines {positions[:20]}{' ...' if len(positions)>20 else ''}"
    )
    report_lines.append(f"  stats: {stats}")
    report_lines.append(f"  blocks: {len(blocks)}")

    # Per-block stats + writing
    base = safe_stem(path.stem) if SANITIZE_OUTPUT_FILENAMES else path.stem
    per_block_stats: List[Dict[str, Any]] = []

    for bi, block in enumerate(blocks, start=1):
        bstat = block_column_stats(block, delim)
        per_block_stats.append(bstat)

        out_path = out_dir / f"{base}__block_{bi:03d}.csv"
        n_rows = write_block_csv(block, out_path, delim)

        # Report line with mismatch_rate
        report_lines.append(
            f"    block_{bi:03d}: rows={n_rows}, cols_mode={bstat['cols_mode']}, "
            f"mismatch_rate={bstat['mismatch_rate']:.3f}, out={out_path.name}"
        )

    # Confidence score
    conf = compute_confidence_score(
        total_lines=total_lines,
        noise_lines=noise_lines,
        header_occurrences=len(positions),
        blocks_count=len(blocks),
        per_block_stats=per_block_stats,
    )
    warn = "  WARNING: low confidence; review split boundaries." if conf["score"] < CONFIDENCE_WARN_THRESHOLD else ""
    report_lines.append(f"  confidence_score: {conf['score']}/100{warn}")
    report_lines.append(f"  confidence_breakdown: {conf['breakdown']}")
    if conf["notes"]:
        report_lines.append(f"  confidence_notes: {conf['notes']}")

    return "\n".join(report_lines)


def main():
    base_dir = get_base_dir()
    out_dir = base_dir / "output_split"
    out_dir.mkdir(parents=True, exist_ok=True)

    targets = sorted([p for p in base_dir.iterdir() if p.is_file() and p.suffix.lower() in (".csv", ".txt")])

    report_path = out_dir / "split_report.txt"
    report_all: List[str] = []
    report_all.append(f"BASE_DIR: {base_dir}")
    report_all.append(f"FOUND_FILES: {len(targets)}")
    report_all.append(f"SANITIZE_OUTPUT_FILENAMES: {SANITIZE_OUTPUT_FILENAMES}")
    report_all.append(f"CONFIDENCE_WARN_THRESHOLD: {CONFIDENCE_WARN_THRESHOLD}")
    report_all.append("")

    if not targets:
        report_all.append("No .csv/.txt files found in the script/exe directory.")
        report_path.write_text("\n".join(report_all), encoding="utf-8")
        print("\n".join(report_all))
        return

    for p in targets:
        rep = process_file(p, out_dir)
        report_all.append(rep)
        report_all.append("-" * 60)

    report_path.write_text("\n".join(report_all), encoding="utf-8")
    print(f"Done. Report: {report_path}")
    print(f"Split CSVs: {out_dir}")


if __name__ == "__main__":
    main()