src/utils.py

"""
APK Deep Diff Analyzer - Utilities Module
"""
import hashlib
import os
import tempfile
import shutil
import zipfile
from pathlib import Path
from typing import Dict, List, Tuple, Optional, Any
import time


def get_file_hash(filepath: str, algorithm: str = "sha256", chunk_size: int = 8192) -> str:
    """Calculate file hash using streaming (memory efficient)."""
    h = hashlib.new(algorithm)
    with open(filepath, "rb") as f:
        while chunk := f.read(chunk_size):
            h.update(chunk)
    return h.hexdigest()


def get_file_size_formatted(size_bytes: int) -> str:
    """Format file size in human readable format."""
    if size_bytes < 1024:
        return f"{size_bytes} B"
    elif size_bytes < 1024 * 1024:
        return f"{size_bytes / 1024:.2f} KB"
    elif size_bytes < 1024 * 1024 * 1024:
        return f"{size_bytes / (1024 * 1024):.2f} MB"
    else:
        return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB"


def create_temp_dir(prefix: str = "apk_") -> str:
    """Create a temporary directory."""
    return tempfile.mkdtemp(prefix=prefix)


def cleanup_temp_dir(path: str):
    """Remove temporary directory and all contents."""
    if os.path.exists(path):
        shutil.rmtree(path, ignore_errors=True)


def extract_apk(apk_path: str, extract_dir: str) -> bool:
    """Extract APK (ZIP) contents to directory."""
    try:
        with zipfile.ZipFile(apk_path, 'r') as zf:
            zf.extractall(extract_dir)
        return True
    except Exception:
        return False


def get_apk_info(apk_path: str) -> Dict[str, Any]:
    """Get basic APK file information."""
    info = {
        "filename": os.path.basename(apk_path),
        "size": os.path.getsize(apk_path),
        "size_formatted": get_file_size_formatted(os.path.getsize(apk_path)),
        "sha256": get_file_hash(apk_path, "sha256"),
        "md5": get_file_hash(apk_path, "md5"),
    }
    return info


def list_files_recursive(directory: str) -> List[Tuple[str, int, str]]:
    """
    List all files recursively.
    Returns list of (relative_path, size, sha256) tuples.
    """
    files = []
    base_path = Path(directory)
    for filepath in base_path.rglob("*"):
        if filepath.is_file():
            rel_path = filepath.relative_to(base_path).as_posix()
            size = filepath.stat().st_size
            sha256 = get_file_hash(str(filepath))
            files.append((rel_path, size, sha256))
    return files


def get_zip_entries(apk_path: str) -> List[Dict[str, Any]]:
    """Get ZIP entry information from APK."""
    entries = []
    try:
        with zipfile.ZipFile(apk_path, 'r') as zf:
            for info in zf.infolist():
                entries.append({
                    "filename": info.filename,
                    "size": info.file_size,
                    "compressed_size": info.compress_size,
                    "compression": info.compress_type,
                    "crc": info.CRC,
                })
    except Exception:
        pass
    return entries


def calculate_similarity(size1: int, size2: int, hash1: str, hash2: str) -> float:
    """Calculate a simple similarity score between two files."""
    if hash1 == hash2:
        return 100.0
    if size1 == 0 and size2 == 0:
        return 100.0
    if size1 == 0 or size2 == 0:
        return 0.0
    size_diff = abs(size1 - size2) / max(size1, size2)
    similarity = max(0.0, (1.0 - size_diff) * 100)
    return round(similarity, 2)


def format_time_elapsed(seconds: float) -> str:
    """Format elapsed time."""
    if seconds < 60:
        return f"{seconds:.2f} ثانية"
    elif seconds < 3600:
        return f"{seconds / 60:.2f} دقيقة"
    else:
        return f"{seconds / 3600:.2f} ساعة"


class ProgressTracker:
    """Track analysis progress."""

    STAGES = [
        "Uploading",
        "Validating APK",
        "Extracting APK",
        "Analyzing ZIP structure",
        "Analyzing AndroidManifest.xml",
        "Comparing resources",
        "Comparing DEX files",
        "Comparing native libraries",
        "Comparing assets",
        "Comparing signatures",
        "Calculating hashes",
        "Generating report",
    ]

    def __init__(self):
        self.current_stage = 0
        self.start_time = time.time()
        self.stage_times = {}

    def set_stage(self, stage_index: int):
        """Set current stage."""
        if 0 <= stage_index < len(self.STAGES):
            self.current_stage = stage_index
            self.stage_times[stage_index] = time.time()

    def get_progress(self) -> Tuple[float, str, str]:
        """Get progress percentage, current stage name, and time elapsed."""
        total = len(self.STAGES)
        pct = (self.current_stage / total) * 100
        stage_name = self.STAGES[self.current_stage] if self.current_stage < total else "Complete"
        elapsed = format_time_elapsed(time.time() - self.start_time)
        return pct, stage_name, elapsed

    def get_stage_time(self, stage_index: int) -> str:
        """Get time spent on a specific stage."""
        if stage_index in self.stage_times and stage_index + 1 in self.stage_times:
            elapsed = self.stage_times[stage_index + 1] - self.stage_times[stage_index]
            return format_time_elapsed(elapsed)
        return "N/A"