source_exporter.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/tools/source_exporter.py
 
"""
Purpose
-------
Provide a unified, Option C‑compliant source‑exporting tool that collects,
combines, and consolidates all source‑code files in a project into structured
inspection files. This enables fast review, debugging, auditing, and
cross‑project comparison.
 
Responsibilities
----------------
- Walk a project directory tree while respecting exclusion rules.
- Collect all source files with supported extensions.
- Generate per‑directory `_source.<path>.txt` files.
- Generate a root‑level `_full_source.<root>.txt` file.
- Support `--clean` to remove all `_source_code-*` directories.
- Support `--verbose` to print per‑file statistics.
- Integrate cleanly with the OCT CLI as `oct export-source`.
- Handle I/O errors (including Windows MAX_PATH) gracefully: warn and continue.
 
Diagnostics
-----------
Domain: OCT-EXPORTER
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Must not modify project files except inside `_source_code-*` directories.
- Must accept any directory as a valid export root.
- Must remain deterministic and safe for large projects.
- Must not require external dependencies beyond the standard library.
- Must never crash on a recoverable I/O error; warn and continue instead.
"""
 
import fnmatch
import hashlib
import json
import os
import re
import sys
import shutil
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Tuple
 
 
# ============================================================
# OI-416: content-level secret scanning
# ============================================================
 
# Compiled once; matches ``name = "value"`` / ``name: "value"`` style
# assignments on a single line. Multi-line heredocs and concatenation are
# intentionally out of scope — this is a "shallow guard" before upload.
_SECRET_ASSIGN_RE = re.compile(
    r'(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*[:=]\s*'
    r'(?:"(?P<dq>[^"\n]{4,})"|\'(?P<sq>[^\'\n]{4,})\')'
)
 
 
def _scan_content_for_secrets(content: str, rel_path: str) -> list[str]:
    """Scan ``content`` for likely hardcoded secrets. Returns warning strings.
 
    Uses the same name-heuristics as the linter (``_SECRET_NAME_PATTERNS``).
    Empty list means no warnings. The ``rel_path`` is embedded in each
    warning for user attribution.
    """
    try:
        # OI-514: scanner moved to a standalone module; source_exporter no
        # longer pulls in the whole linter just to reach the heuristics.
        from oct.tools.secret_scanner import _name_matches_secret
    except Exception:
        return []
 
    warnings: list[str] = []
    for lineno, line in enumerate(content.splitlines(), 1):
        for m in _SECRET_ASSIGN_RE.finditer(line):
            name = m.group("name")
            if _name_matches_secret(name):
                warnings.append(
                    f"{rel_path}:{lineno}: potential secret '{name}'"
                )
                break  # one warning per line is enough
    return warnings
 
 
# Cache of per-file secret scan results, keyed by absolute path. Used to
# dedup warnings between the per-directory and full-project passes while
# still allowing both passes to make block decisions.
_secret_scan_cache: dict[str, list[str]] = {}
# Paths whose warnings have already been emitted to stderr this run.
_scanned_secret_paths: set[str] = set()
 
 
def _get_cached_secret_warnings(abs_path: str, content: str, rel_path: str) -> list[str]:
    """Return (cached or computed) secret warnings for ``abs_path``."""
    if abs_path not in _secret_scan_cache:
        _secret_scan_cache[abs_path] = _scan_content_for_secrets(content, rel_path)
    return _secret_scan_cache[abs_path]
 
 
# ============================================================
# Configuration
# ============================================================
 
_EXTENSION_PROFILES = {
    "python": [".py"],
    "web": [".js", ".ts", ".tsx", ".css", ".html"],
    "jvm": [".java", ".kt"],
    "systems": [".rs", ".go", ".cpp", ".c", ".h", ".hpp"],
    "config": [".json", ".yaml", ".yml", ".toml", ".ini", ".xml"],
    "docs": [".md"],
    "scripts": [".sh", ".bat", ".ps1", ".ahk"],
    "misc": [".npmignore", ".gitignore", ".sql", ".rb", ".swift"],
}
 
INCLUDE_EXTENSIONS = [ext for exts in _EXTENSION_PROFILES.values() for ext in exts]
 
from oct.core.exclusions import EXPORTER_EXCLUDE_DIRS as EXCLUDE_DIRS
from oct.core.exclusions import WILDCARD_EXCLUDE_DIRS
from oct.core.exclusions import NEVER_EXPORT_FILE_PATTERNS
 
HEADER_TEXT = (
    "This file contains all source-code files in the directory {directory} "
    "(and its subdirectories) of this project.\n"
    "Sections are separated as follows:\n"
    "- '##### File name XX: <path>' marks individual files.\n"
    "- '####### Subdirectory name XX: <dirname>' marks combined subdirectory sources.\n"
    "- Root-level files use './<filename>' for clarity.\n\n"
)
 
SEPARATOR_FILE = "##### File name {num:02d}: {path}\n"
SEPARATOR_SUBDIR = "####### Subdirectory name {num:02d}: {dirname}\n"
ROOT_HEADER = "####### Root directory: .\n\n"
 
# Windows MAX_PATH guard: keep well below the 260-character hard limit
_MAX_SAFE_PATH_LEN = 240
 
_CONFIG_FILENAME = "_oct_exporter_config.json"
 
 
# ============================================================
# Helpers
# ============================================================
 
def load_exporter_config(root: Path) -> dict:
    """Load optional per-project config, merging with module-level defaults."""
    cfg = {
        "include_extensions": list(INCLUDE_EXTENSIONS),
        "exclude_dirs": set(EXCLUDE_DIRS),
        "wildcard_exclude_dirs": list(WILDCARD_EXCLUDE_DIRS),
        "header_text": HEADER_TEXT,
        "separator_file": SEPARATOR_FILE,
        "separator_subdir": SEPARATOR_SUBDIR,
        "root_header": ROOT_HEADER,
        "max_safe_path_len": _MAX_SAFE_PATH_LEN,
    }
 
    config_path = root / _CONFIG_FILENAME
    if not config_path.is_file():
        # Fall back to oct's own bundled config as the default.
        _oct_root = Path(__file__).resolve().parent.parent.parent
        fallback_path = _oct_root / _CONFIG_FILENAME
        if fallback_path.is_file() and fallback_path != config_path:
            print(f"  Note: No {_CONFIG_FILENAME} in {root}; "
                  f"using oct default from {fallback_path}.",
                  file=sys.stderr)
            config_path = fallback_path
        else:
            print(f"  Warning: No {_CONFIG_FILENAME} found; using built-in defaults.",
                  file=sys.stderr)
            return cfg
 
    try:
        raw = json.loads(config_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as e:
        print(f"  Warning: {_CONFIG_FILENAME} has invalid JSON ({e}); using defaults.",
              file=sys.stderr)
        return cfg
 
    if not isinstance(raw, dict):
        print(f"  Warning: {_CONFIG_FILENAME} root must be a JSON object; using defaults.",
              file=sys.stderr)
        return cfg
 
    print(f"Using {_CONFIG_FILENAME}")
 
    def _merge_collection(key, default, collection_type):
        """Apply replace / add / remove semantics for a collection-type key."""
        add_key = f"{key}_add"
        remove_key = f"{key}_remove"
 
        if key in raw:
            result = collection_type(raw[key])
            print(f"  {key}: replaced -> {sorted(result)}")
        else:
            result = collection_type(default)
 
        if add_key in raw:
            added = raw[add_key]
            if collection_type is set:
                result = result | set(added)
            else:
                result = result + [v for v in added if v not in result]
            print(f"  {key}: added {added}")
 
        if remove_key in raw:
            removed = raw[remove_key]
            removals = set(removed)
            if collection_type is set:
                result = result - removals
            else:
                result = [v for v in result if v not in removals]
            print(f"  {key}: removed {removed}")
 
        return result
 
    cfg["include_extensions"] = _merge_collection(
        "include_extensions", INCLUDE_EXTENSIONS, list
    )
    cfg["exclude_dirs"] = _merge_collection(
        "exclude_dirs", EXCLUDE_DIRS, set
    )
    cfg["wildcard_exclude_dirs"] = _merge_collection(
        "wildcard_exclude_dirs", WILDCARD_EXCLUDE_DIRS, list
    )
 
    for scalar_key in ("header_text", "separator_file", "separator_subdir",
                       "root_header", "max_safe_path_len"):
        if scalar_key in raw:
            cfg[scalar_key] = raw[scalar_key]
            print(f"  {scalar_key}: overridden -> {raw[scalar_key]!r}")
 
    return cfg
 
 
def should_skip_dirname(dirname: str, cfg: dict) -> bool:
    name = dirname.lower()
    if name in cfg["exclude_dirs"]:
        return True
    return any(pattern in name for pattern in cfg["wildcard_exclude_dirs"])
 
 
def _is_never_export_file(filename: str) -> bool:
    """Check if a file matches NEVER_EXPORT_FILE_PATTERNS (§6b Secret Hygiene)."""
    name_lower = filename.lower()
    for pattern in NEVER_EXPORT_FILE_PATTERNS:
        if fnmatch.fnmatch(name_lower, pattern.lower()):
            return True
    return False
 
 
def collect_recursive_files(folder: Path, cfg: dict) -> List[Path]:
    collected = []
    for root, dirs, files in os.walk(folder):
        root_path = Path(root)
        if should_skip_dirname(root_path.name, cfg):
            dirs[:] = []
            continue
        dirs[:] = [d for d in dirs if not should_skip_dirname(d, cfg)]
        for f in files:
            if _is_never_export_file(f):
                continue
            if any(f.endswith(ext) for ext in cfg["include_extensions"]):
                collected.append(root_path / f)
    return sorted(collected)
 
 
def get_file_stats(path: Path) -> Tuple[int, int, int]:
    try:
        content = path.read_text(encoding="utf-8", errors="replace")
    except (OSError, UnicodeDecodeError):
        print(f"Warning: Could not read {path}: skipping", file=sys.stderr)
        return 0, 0, 0
    lines = content.count("\n") + 1 if content else 0
    words = len(content.split()) if content else 0
    try:
        bytes_ = path.stat().st_size
    except OSError:
        bytes_ = len(content.encode("utf-8"))
    return lines, words, bytes_
 
 
def walk_included_dirs(start: Path, cfg: dict):
    for root, dirs, _ in os.walk(start, topdown=True):
        root_path = Path(root)
        if should_skip_dirname(root_path.name, cfg):
            dirs[:] = []
            continue
        dirs[:] = [d for d in dirs if not should_skip_dirname(d, cfg)]
        yield root_path
 
 
def clean_source_dirs(root: Path, dry_run: bool = False) -> None:
    removed = 0
    for dirpath, dirs, _ in os.walk(root, topdown=True):
        for d in list(dirs):
            if d.startswith("_source_code-"):
                target = Path(dirpath) / d
                if dry_run:
                    print(f"[DRY RUN] Would remove: {target}")
                else:
                    shutil.rmtree(target, ignore_errors=True)
                dirs.remove(d)
                removed += 1
    if dry_run:
        print(f"[DRY RUN] Would remove {removed} _source_code-* directories under {root}")
    else:
        print(f"Removed {removed} _source_code-* directories under {root}")
 
 
def clean_pycache_dirs(root: Path, dry_run: bool = False) -> None:
    """Remove all __pycache__ directories under root."""
    removed = 0
    for dirpath, dirs, _ in os.walk(root, topdown=True):
        for d in list(dirs):
            if d == "__pycache__":
                target = Path(dirpath) / d
                if dry_run:
                    print(f"[DRY RUN] Would remove: {target}")
                else:
                    shutil.rmtree(target, ignore_errors=True)
                dirs.remove(d)
                removed += 1
    if dry_run:
        print(f"[DRY RUN] Would remove {removed} __pycache__ directories under {root}")
    else:
        print(f"Removed {removed} __pycache__ directories under {root}")
 
 
def _make_safe_output_path(out_dir: Path, dotpath: str, cfg: dict,
                           prefix: str = "_source") -> Path:
    """
    Build the output file path, truncating the dotpath if the full path would
    exceed the Windows MAX_PATH limit (~260 chars). A short MD5 hash suffix is
    appended when truncation is applied to ensure uniqueness.
    """
    max_len = cfg["max_safe_path_len"]
    filename = f"{prefix}.{dotpath}.txt"
    candidate = out_dir / filename
    if len(str(candidate)) <= max_len:
        return candidate
    h = hashlib.md5(dotpath.encode()).hexdigest()[:8]
    overhead = len(str(out_dir)) + len(f"{prefix}....{h}.txt") + 2  # 2 for sep + slack
    max_dotpath_len = max_len - overhead
    truncated = dotpath[:max(0, max_dotpath_len)].rstrip(".")
    return out_dir / f"{prefix}.{truncated}...{h}.txt"
 
 
# ============================================================
# Core exporter logic
# ============================================================
 
def write_source_file_for_dir(folder: Path, output_path: Path, root: Path,
                              cfg: dict, warn_secrets: bool = False,
                              block_secrets: bool = False,
                              diff_filter: set[Path] | None = None):
    files = collect_recursive_files(folder, cfg)
    if diff_filter is not None:
        files = [f for f in files if f.resolve() in diff_filter]
        if not files:
            return 0, 0, 0, 0, []
 
    file_count = 0
    total_lines = 0
    total_words = 0
    total_bytes = 0
 
    try:
        fh = output_path.open("w", encoding="utf-8")
    except OSError as e:
        print(f"  Warning: cannot write '{output_path.name}': {e}")
        return 0, 0, 0, 0, files
 
    with fh as out:
        out.write(cfg["header_text"].format(directory=folder))
        out.write(cfg["root_header"])
 
        for idx, file_path in enumerate(files, start=1):
            rel_path = file_path.relative_to(folder)
 
            try:
                content = file_path.read_text(encoding="utf-8", errors="replace")
            except (OSError, UnicodeDecodeError):
                print(f"Warning: Could not read {file_path}: skipping", file=sys.stderr)
                content = ""
 
            # OI-416: content-level secret scanning
            if warn_secrets and content:
                abs_key = str(file_path.resolve())
                rel_from_root = (
                    file_path.relative_to(root)
                    if file_path.is_relative_to(root) else rel_path
                )
                sec_warnings = _get_cached_secret_warnings(
                    abs_key, content, str(rel_from_root)
                )
                if abs_key not in _scanned_secret_paths:
                    _scanned_secret_paths.add(abs_key)
                    for w in sec_warnings:
                        print(f"Warning: {w}", file=sys.stderr)
                if block_secrets and sec_warnings:
                    print(
                        f"  Skipping {rel_from_root} due to potential secrets",
                        file=sys.stderr,
                    )
                    continue
 
            out.write(cfg["separator_file"].format(num=idx, path=str(rel_path)))
            out.write(content)
            out.write("\n\n")
 
            file_count += 1
            total_bytes += file_path.stat().st_size
            if content:
                total_lines += content.count("\n") + 1
                total_words += len(content.split())
 
    return file_count, total_lines, total_words, total_bytes, files
 
 
# ============================================================
# Public entry point used by OCT CLI
# ============================================================
 
def run_exporter(root: Path, args: list[str]) -> None:
    """
    Execute the source exporter with the given root directory and arguments.
    """
    do_clean = "--clean" in args
    verbose = "--verbose" in args
    single_dir_mode = "--single-dir" in args
    warn_secrets = "--warn-secrets" in args or "--block-secrets" in args
    block_secrets = "--block-secrets" in args
 
    # Phase 4E — G-E6: --diff REF support.
    diff_ref: str | None = None
    for arg in args:
        if arg.startswith("--diff="):
            diff_ref = arg.split("=", 1)[1]
            break
 
    # OI-416: reset scan caches at the start of each run so repeated
    # invocations (e.g. from tests) emit warnings independently.
    _scanned_secret_paths.clear()
    _secret_scan_cache.clear()
 
    if do_clean:
        clean_source_dirs(root)
        return
 
    # Pre-compute diff filter set when --diff is active.
    diff_filter: set[Path] | None = None
    if diff_ref:
        try:
            from oct.core.git import git_diff_name_only
            diff_paths = git_diff_name_only(root, diff_ref)
            diff_filter = set(diff_paths)
            if not diff_filter:
                print(f"No files changed since {diff_ref}. Nothing to export.", file=sys.stderr)
                return
            print(f"Exporting {len(diff_filter)} file(s) changed since {diff_ref}.", file=sys.stderr)
        except Exception as exc:
            print(f"Warning: --diff failed ({exc}). Exporting all files.", file=sys.stderr)
 
    cfg = load_exporter_config(root)
 
    timestamp = datetime.now().strftime("%Y%m%d-%H%M")
    root_name = root.name
 
    summary: Dict[str, Tuple[int, int, int, int]] = {}
    per_dir_files: Dict[str, List[Path]] = {}
    skipped: List[str] = []
 
    # In single-dir mode, all output goes under one root-level directory
    single_out_root: Path | None = None
    if single_dir_mode:
        single_out_root = root / f"_source_code-{timestamp}"
        try:
            single_out_root.mkdir(parents=True, exist_ok=True)
        except OSError as e:
            print(f"  Error: cannot create single output dir: {e}")
            return
 
    # Per-directory exports
    for folder in walk_included_dirs(root, cfg):
        if should_skip_dirname(folder.name, cfg):
            continue
 
        if single_dir_mode:
            rel_to_root = folder.relative_to(root) if folder != root else Path(".")
            out_dir = single_out_root / rel_to_root
        else:
            out_dir = folder / f"_source_code-{timestamp}"
 
        try:
            out_dir.mkdir(parents=True, exist_ok=True)
        except OSError as e:
            rel_err = "." if folder == root else str(folder.relative_to(root))
            print(f"  Warning: cannot create output dir for '{folder.name}': {e}")
            skipped.append(rel_err)
            continue
 
        rel = "." if folder == root else str(folder.relative_to(root))
        dotpath = root_name if folder == root else f"{root_name}.{rel.replace(os.sep, '.')}"
        out_path = _make_safe_output_path(out_dir, dotpath, cfg)
 
        try:
            file_count, total_lines, total_words, total_bytes, files = write_source_file_for_dir(
                folder, out_path, root, cfg,
                warn_secrets=warn_secrets, block_secrets=block_secrets,
                diff_filter=diff_filter,
            )
        except Exception as e:
            print(f"  Warning: export failed for '{rel}': {e}")
            skipped.append(rel)
            continue
 
        summary[rel] = (file_count, total_lines, total_words, total_bytes)
        per_dir_files[rel] = files
 
    # Full project export
    if single_dir_mode:
        root_out_dir = single_out_root
    else:
        root_out_dir = root / f"_source_code-{timestamp}"
    full_output_path = _make_safe_output_path(root_out_dir, root_name, cfg,
                                              prefix="_full_source")
 
    all_files = collect_recursive_files(root, cfg)
    if diff_filter is not None:
        all_files = [f for f in all_files if f.resolve() in diff_filter]
    grouped: Dict[str, List[Path]] = {}
    for p in all_files:
        parent_rel = p.parent.relative_to(root)
        key = "." if str(parent_rel) == "." else str(parent_rel)
        grouped.setdefault(key, []).append(p)
 
    try:
        fh = full_output_path.open("w", encoding="utf-8")
    except OSError as e:
        print(f"  Warning: cannot write full-source file: {e}")
        fh = None
 
    if fh is not None:
        with fh as out:
            out.write(cfg["header_text"].format(directory=root))
            out.write(cfg["root_header"])
 
            for idx_dir, dirname in enumerate(sorted(grouped.keys()), start=1):
                out.write(cfg["separator_subdir"].format(num=idx_dir, dirname=dirname))
                for idx_file, file_path in enumerate(grouped[dirname], start=1):
                    rel_path = file_path.relative_to(root)
                    try:
                        content = file_path.read_text(encoding="utf-8", errors="replace")
                    except (OSError, UnicodeDecodeError):
                        print(f"Warning: Could not read {file_path}: skipping", file=sys.stderr)
                        content = ""
 
                    # OI-416: content-level secret scanning (dedup with per-dir pass).
                    if warn_secrets and content:
                        abs_key = str(file_path.resolve())
                        sec_warnings = _get_cached_secret_warnings(
                            abs_key, content, str(rel_path)
                        )
                        if abs_key not in _scanned_secret_paths:
                            _scanned_secret_paths.add(abs_key)
                            for w in sec_warnings:
                                print(f"Warning: {w}", file=sys.stderr)
                        if block_secrets and sec_warnings:
                            print(
                                f"  Skipping {rel_path} due to potential secrets",
                                file=sys.stderr,
                            )
                            continue
 
                    out.write(cfg["separator_file"].format(num=idx_file, path=str(rel_path)))
                    out.write(content)
                    out.write("\n\n")
 
    # Summary
    print("\n========== SUMMARY ==========")
 
    seen_files = set()
 
    for dirname in sorted(summary.keys()):
        files, lines, words, bytes_ = summary[dirname]
        kb = bytes_ / 1024.0
        print(f"Directory '{dirname}': {files} files, {lines} lines, {words} words, {kb:.1f} KB")
 
        if verbose:
            for p in per_dir_files.get(dirname, []):
                if p not in seen_files:
                    seen_files.add(p)
                    flines, fwords, fbytes = get_file_stats(p)
                    fkb = fbytes / 1024.0
                    rel_path = p.relative_to(root)
                    print(f"    {rel_path} — {flines} lines, {fwords} words, {fkb:.1f} KB")
 
    # Grand total: each per-directory entry counts files recursively, so summing
    # them all would multiply each file by the number of ancestor directories it
    # has. Use the root entry instead — it already covers the full project.
    # Fall back to counting unique files directly if root was skipped due to error.
    if "." in summary:
        total_files, total_lines, total_words, total_bytes = summary["."]
    else:
        total_files = total_lines = total_words = total_bytes = 0
        for p in all_files:
            flines, fwords, fbytes = get_file_stats(p)
            total_files += 1
            total_lines += flines
            total_words += fwords
            total_bytes += fbytes
 
    print(f"\nTotal directories processed: {len(summary)}")
    print(f"Total source files processed: {total_files}")
    print(f"Total lines: {total_lines}")
    print(f"Total words: {total_words}")
    print(f"Total size: {total_bytes / 1024.0:.1f} KB")
    if single_dir_mode:
        print(f"Output stored in single directory: {single_out_root}")
    else:
        print(f"Output stored under _source_code-{timestamp} directories")
 
    if skipped:
        n = len(skipped)
        print(f"\nSkipped {n} director{'y' if n == 1 else 'ies'} due to errors:")
        for s in skipped:
            print(f"  - {s}")
 
    print("=============================\n")