# main_analyzer.py import fitz # PyMuPDF import os import tempfile import re import traceback from typing import Tuple, Dict, Any, List from collections import defaultdict from pdf_processing import ( extract_font_filtered_markdown, extract_plain_text_from_original_pdf, try_map_issues_to_page_rects ) from content_analysis import ( check_metadata, check_disclosures, check_figures_and_tables, check_references_summary, check_structure, check_figure_order, check_reference_order ) from language_checker import perform_language_checks from regex_checker import perform_regex_checks def analyze_pdf(filepath_or_stream: Any) -> Tuple[Dict[str, Any], None]: original_pdf_access_path = None temp_file_for_stream_path = None doc_for_mapping = None try: if isinstance(filepath_or_stream, str): original_pdf_access_path = filepath_or_stream print(f"Analyzer: Input is a string path: {original_pdf_access_path}") # Check for objects like Gradio's NamedString or TemporaryFileWrapper's .name attribute elif hasattr(filepath_or_stream, 'name') and isinstance(getattr(filepath_or_stream, 'name'), str) and \ os.path.exists(getattr(filepath_or_stream, 'name')): # Ensure the .name path is valid original_pdf_access_path = filepath_or_stream.name print(f"Analyzer: Input is an object with .name attribute, using path: {original_pdf_access_path}") # If this object also has a .read method, it might be a TemporaryFileWrapper. # The next elif would handle it if we prefer processing it as a stream, # but using its .name path is usually fine and simpler. elif hasattr(filepath_or_stream, 'read') and callable(filepath_or_stream.read): with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as temp_file_obj: temp_file_for_stream_path = temp_file_obj.name if hasattr(filepath_or_stream, 'seek') and callable(filepath_or_stream.seek): filepath_or_stream.seek(0) temp_file_obj.write(filepath_or_stream.read()) original_pdf_access_path = temp_file_for_stream_path print(f"Analyzer: Input stream saved to temp file: {original_pdf_access_path}") else: return {"error": f"Invalid PDF input type: {type(filepath_or_stream)}. Must be path string, an object with a .name attribute as path, or file-like stream object."}, None if not original_pdf_access_path or not os.path.exists(original_pdf_access_path): return {"error": f"PDF path '{original_pdf_access_path}' (derived from input) does not exist or is invalid."}, None # --- The rest of the function remains the same as the previous complete listing --- # 1. Unfiltered Plain Text (for general and regex checks) print(f"Analyzer: Extracting plain text from original PDF: {original_pdf_access_path}") raw_unfiltered_plain_text = extract_plain_text_from_original_pdf(original_pdf_access_path) pdf_size = os.path.getsize(original_pdf_access_path) if not raw_unfiltered_plain_text and pdf_size > 0 : print("Analyzer: Warning: Raw unfiltered plain text extraction yielded empty result. PDF might be image-based or have extraction issues.") cleaned_unfiltered_plain_text = re.sub(r'\s+', ' ', raw_unfiltered_plain_text.replace('\n', ' ')).strip() # 2. Font-Filtered Markdown (for LanguageTool checks) print(f"Analyzer: Extracting font-filtered markdown from: {original_pdf_access_path}") markdown_text_from_filtered_pdf = extract_font_filtered_markdown(original_pdf_access_path) if not markdown_text_from_filtered_pdf and pdf_size > 0 : print("Analyzer: Warning: Font-filtered Markdown extraction yielded empty result.") # 3. Perform all checks document_check_results = { "metadata": check_metadata(cleaned_unfiltered_plain_text), "disclosures": check_disclosures(cleaned_unfiltered_plain_text), "figures_and_tables": check_figures_and_tables(cleaned_unfiltered_plain_text), "references_summary": check_references_summary(cleaned_unfiltered_plain_text), "structure": check_structure(cleaned_unfiltered_plain_text), "figure_order_analysis": check_figure_order(cleaned_unfiltered_plain_text), "reference_order_analysis": check_reference_order(cleaned_unfiltered_plain_text), "plain_language_summary_present": bool(re.search(r'plain language summary', cleaned_unfiltered_plain_text, re.IGNORECASE)), "readability_issues_detected": False, } print("Analyzer: Performing regex checks...") regex_report = perform_regex_checks(cleaned_unfiltered_plain_text) if "error" in regex_report: print(f"Analyzer: Error in regex checks: {regex_report['error']}") regex_issues = regex_report.get("issues_list", []) print("Analyzer: Performing language checks...") lt_report = perform_language_checks(markdown_text_from_filtered_pdf) if "error" in lt_report: print(f"Analyzer: Error in LanguageTool checks: {lt_report['error']}") lt_issues = lt_report.get("issues_list", []) detailed_issues_for_mapping = regex_issues + lt_issues # 4. Coordinate Mapping (against the original PDF) if detailed_issues_for_mapping: try: doc_for_mapping = fitz.open(original_pdf_access_path) if doc_for_mapping.page_count > 0: print(f"Analyzer: Mapping {len(detailed_issues_for_mapping)} issues to PDF coordinates...") for page_idx in range(doc_for_mapping.page_count): page = doc_for_mapping[page_idx] current_page_num_1_based = page_idx + 1 unmapped_issues_on_this_page_by_context = defaultdict(list) for issue_dict in detailed_issues_for_mapping: if not issue_dict['is_mapped_to_pdf']: unmapped_issues_on_this_page_by_context[issue_dict['context_text']].append(issue_dict) if not unmapped_issues_on_this_page_by_context: if all(iss['is_mapped_to_pdf'] for iss in detailed_issues_for_mapping): break continue for ctx_str, issues_for_ctx in unmapped_issues_on_this_page_by_context.items(): if not ctx_str or not ctx_str.strip(): continue try: pdf_rects = page.search_for(ctx_str, flags=fitz.TEXT_PRESERVE_LIGATURES | fitz.TEXT_PRESERVE_WHITESPACE) if pdf_rects: try_map_issues_to_page_rects(issues_for_ctx, pdf_rects, current_page_num_1_based) except Exception as search_exc: print(f"Analyzer: Warning: Error searching for context '{ctx_str[:30].replace(chr(10),' ')}' on page {current_page_num_1_based}: {search_exc}") total_mapped = sum(1 for iss in detailed_issues_for_mapping if iss['is_mapped_to_pdf']) print(f"Analyzer: Finished coordinate mapping. Mapped issues: {total_mapped}/{len(detailed_issues_for_mapping)}.") except Exception as e_map: print(f"Analyzer: Error during PDF coordinate mapping: {e_map}\n{traceback.format_exc()}") finally: if doc_for_mapping: doc_for_mapping.close() else: print("Analyzer: No detailed issues from regex or language checks to map.") # 5. Format final list of issues final_formatted_issues_list = [] for issue_data in detailed_issues_for_mapping: coords = issue_data.get('pdf_coordinates_list', [{}])[0] if issue_data.get('is_mapped_to_pdf') else {} coords_for_json = [coords.get("x0"), coords.get("y0"), coords.get("x1"), coords.get("y1")] if coords else [] coords_for_json = [c for c in coords_for_json if c is not None] final_formatted_issues_list.append({ "message": issue_data.get('message', 'N/A'), "context": issue_data.get('context_text', 'N/A'), "suggestions": issue_data.get('replacements_suggestion', []), "category": issue_data.get('category_name', 'Unknown'), "rule_id": issue_data.get('ruleId', 'N/A'), "offset": issue_data.get('offset_in_text', -1), "length": issue_data.get('error_length', 0), "coordinates": coords_for_json if len(coords_for_json) == 4 else [], "page": issue_data.get('mapped_page_number', 0) if issue_data.get('is_mapped_to_pdf') else 0, "source_check_type": issue_data.get('source_check_type', 'N/A') }) results = { "issues": final_formatted_issues_list, "document_checks": document_check_results } return results, None except Exception as e: print(f"Overall analysis error in analyze_pdf: {e}\n{traceback.format_exc()}") return {"error": f"Overall analysis error: {str(e)}"}, None finally: if temp_file_for_stream_path and os.path.exists(temp_file_for_stream_path): try: os.remove(temp_file_for_stream_path) print(f"Analyzer: Cleaned up main temporary PDF file: {temp_file_for_stream_path}") except Exception as e_clean: print(f"Analyzer: Error cleaning up main temporary PDF file {temp_file_for_stream_path}: {e_clean}")