pip install pymupdf

import fitz

def get_pdf_content_from_path(pdf_path):
    """Get PDF content from a local file path"""
    pdf = fitz.open(pdf_path)
    return pdf 

def get_pdf_content_from_datalake(content_url):
    """Get PDF content using content_url"""
    content = get_content_by_content_url(content_url)
    try:
        pdf = fitz.open(filetype="pdf", stream=content)
    except Exception as e:
        raise ValueError(f"Failed to open PDF from DataLake for URL: {content_url}. Error: {str(e)}")
    return pdf

  # Determine the maximum number of pages
    max_pages = max(len(pdf_base), len(pdf_target))

def extract_page_data(pdf, page_num):
    """Extract text and pixel data from a PDF page."""
    page = pdf[page_num]
    text = page.get_text()
    pix = page.get_pixmap()
    return {
        "text": text,
        "pix_samples": pix.samples,
        "pix_width": pix.width,
        "pix_height": pix.height,
    }
def generate_page_data(pdf_base, pdf_target, max_pages, doc_folder):
    """Generator to yield page data for multiprocessing."""
    for page_num in range(max_pages):
        page_data_base = extract_page_data(pdf_base, page_num)
        page_data_target = extract_page_data(pdf_target, page_num)
        yield (page_data_base, page_data_target, page_num, doc_folder)

compare_metadata(pdf_base.metadata, pdf_target.metadata, result)
def compare_metadata(metadata_base, metadata_target, result):
    """Compare PDF metadata"""
    for key in set(metadata_base.keys()) | set(metadata_target.keys()):
        if metadata_base.get(key) != metadata_target.get(key):
            result["metadata_differences"].append(
                f"Metadata '{key}' differs: pdf_base='{metadata_base.get(key)}', pdf_target='{metadata_target.get(key)}'"
            )

def compare_text_content(page_data_base, page_data_target, page_num, result):
    """Compare text content of two pages."""
    text_base = page_data_base["text"]
    text_target = page_data_target["text"]

    if text_base != text_target:
        result["text_differences"].append(f"Text differs on page {page_num + 1}")
        diff = list(difflib.ndiff(text_base.splitlines(), text_target.splitlines()))
        differences = [d for d in diff if d.startswith('+ ') or d.startswith('- ')]
        if differences:
            result["text_differences"].append(f"Page {page_num + 1} specific differences: {differences[:5]}...")

def compare_visual_content(page_data_base, page_data_target, page_num, doc_folder, result):
    """Compare visual content of two pages."""
    if (page_data_base["pix_width"] != page_data_target["pix_width"] or
            page_data_base["pix_height"] != page_data_target["pix_height"]):
        result["format_differences"].append(
            f"Page {page_num + 1} size differs: PDF_base={page_data_base['pix_width']}x{page_data_base['pix_height']}, "
            f"PDF_target={page_data_target['pix_width']}x{page_data_target['pix_height']}"
        )
        return

    img_base = Image.frombytes("RGB", [page_data_base["pix_width"], page_data_base["pix_height"]], page_data_base["pix_samples"])
    img_target = Image.frombytes("RGB", [page_data_target["pix_width"], page_data_target["pix_height"]], page_data_target["pix_samples"])
    diff_img = ImageChops.difference(img_base, img_target)

    if np.any(np.array(diff_img)):
        img_base_path = os.path.join(doc_folder, f"page_{page_num + 1}_pdf_base.png")
        img_target_path = os.path.join(doc_folder, f"page_{page_num + 1}_pdf_target.png")
        diff_path = os.path.join(doc_folder, f"page_{page_num + 1}_diff.png")
        img_base.save(img_base_path)
        img_target.save(img_target_path)
        diff_img.save(diff_path)
        result["format_differences"].append(f"differs on page {page_num + 1}: difference image saved at {diff_path}")

def hash_page_content(page_data):
    """Generate a hash for the page content."""
    text_hash = hashlib.md5(page_data["text"].encode()).hexdigest()
    pix_hash = hashlib.md5(page_data["pix_samples"]).hexdigest()
    return text_hash, pix_hash


def compare_page(page_data_base, page_data_target, page_num, doc_folder):
    """Compare a single page for text and visual differences."""
    result = {
        "text_differences": [],
        "format_differences": []
    }
    try:
        # Compare hashes first
        base_hash = hash_page_content(page_data_base)
        target_hash = hash_page_content(page_data_target)
        if base_hash == target_hash:
            return result  # Skip comparison if hashes are identical
        
        # Compare text and visual content
        compare_text_content(page_data_base, page_data_target, page_num, result)
        compare_visual_content(page_data_base, page_data_target, page_num, doc_folder, result)

    except Exception as e:
        result["format_differences"].append(f"Failed to compare page {page_num + 1}: {str(e)}")
    
    return result

def compare_page_with_limit(args, diff_page_count, max_diff_pages, lock):
    """Compare a single page with early termination."""
    page_data_base, page_data_target, page_num, doc_folder = args
    with lock:
        if diff_page_count.value >= max_diff_pages:
            return None  # Skip further processing if limit is reached
    page_result = compare_page(page_data_base, page_data_target, page_num, doc_folder)
    if page_result["text_differences"] or page_result["format_differences"]:
        with lock:
            diff_page_count.value += 1
    return page_result

def prepare_output_folder(output_folder, pdf_object_id):
    """Prepare the output folder for storing comparison results."""
    output_folder = os.path.join(constants.OUTPUT_DIR, output_folder)
    os.makedirs(output_folder, exist_ok=True)
    doc_folder = os.path.join(output_folder, pdf_object_id.replace(":", "_"))
    clear_and_create_content_dir(doc_folder)
    return doc_folder


def compare_pdf(pdf_base_path, pdf_target_path, 
                 pdf_object_id, pdf_base_object_url, pdf_target_object_url,
                 is_from_datalake=True, output_folder="pdf_diff_results", 
                 max_diff_pages=3):
    """Compare two PDF files for content and format differences"""
   # Prepare output folder
    doc_folder = prepare_output_folder(output_folder, pdf_object_id)
    
    # Initialize result
    result = {
        "text_differences": [],
        "format_differences": [],
        "metadata_differences": [],
        "page_count": {"pdf_base": 0, "pdf_target": 0}
    }
    
 
    # Open PDF files
    pdf_base = get_pdf_content_from_datalake(pdf_base_object_url) if is_from_datalake else get_pdf_content_from_path(pdf_base_path)
    pdf_target = get_pdf_content_from_datalake(pdf_target_object_url) if is_from_datalake else get_pdf_content_from_path(pdf_target_path)

    
    # Compare page count
    result["page_count"]["pdf_base"] = len(pdf_base)
    result["page_count"]["pdf_target"] = len(pdf_target)
      
    # Compare metadata, ignore differences in creation/modification dates
    # compare_metadata(pdf_base.metadata, pdf_target.metadata, result)
    
     # Determine the maximum number of pages
    max_pages = max(len(pdf_base), len(pdf_target))

    # Compare pages in parallel using a generator
    with Manager() as manager:
        # Shared counter for tracking pages with differences
        diff_page_count = manager.Value('i', 0)
        lock = manager.Lock()
        # Create a pool of worker processes
        with Pool() as pool:
            page_results = pool.starmap(
                compare_page_with_limit,
                [(args, diff_page_count, max_diff_pages, lock) for args in generate_page_data(pdf_base, pdf_target, max_pages, doc_folder)]
            )
    
        if diff_page_count.value >= max_diff_pages:
            print(f"Early termination: {diff_page_count.value} pages with differences found, stopping further processing.")
            pool.terminate()
            pool.join()

    # Aggregate results
    for page_result in page_results:
        if page_result is None:
            continue  # Skip if terminated early
        result["text_differences"].extend(page_result["text_differences"])
        result["format_differences"].extend(page_result["format_differences"])

    return result