diff --git a/.gitignore b/.gitignore index d5cd22cf2..c64fb5c1b 100644 --- a/.gitignore +++ b/.gitignore @@ -94,13 +94,3 @@ yarn-debug.log* .vscode .claude/settings.local.json .cursor - -bin/ -local/ -cache/ -gems/ -specifications/ -extensions/ -build_info/ -bin -.pnpm-store/ diff --git a/db/migrate/20250702075600_add_source_metadata_to_captain_documents.rb b/db/migrate/20250702075600_add_source_metadata_to_captain_documents.rb new file mode 100644 index 000000000..78eb05c3e --- /dev/null +++ b/db/migrate/20250702075600_add_source_metadata_to_captain_documents.rb @@ -0,0 +1,11 @@ +class AddSourceMetadataToCaptainDocuments < ActiveRecord::Migration[7.1] + def change + add_column :captain_documents, :source_type, :string, default: 'url' + add_column :captain_documents, :content_type, :string + add_column :captain_documents, :file_size, :integer + add_column :captain_documents, :processed_at, :datetime + + add_index :captain_documents, :source_type + add_index :captain_documents, :content_type + end +end \ No newline at end of file diff --git a/db/schema.rb b/db/schema.rb index d065a8ff4..79a1ae048 100644 --- a/db/schema.rb +++ b/db/schema.rb @@ -10,7 +10,7 @@ # # It's strongly recommended that you check this file into your version control system. -ActiveRecord::Schema[7.1].define(version: 2025_05_23_031839) do +ActiveRecord::Schema[7.1].define(version: 2025_07_02_075600) do # These extensions should be enabled to support this database enable_extension "pg_stat_statements" enable_extension "pg_trgm" @@ -289,9 +289,17 @@ ActiveRecord::Schema[7.1].define(version: 2025_05_23_031839) do t.datetime "created_at", null: false t.datetime "updated_at", null: false t.integer "status", default: 0, null: false + t.integer "document_type", default: 0, null: false + t.string "source_type", default: "url" + t.string "content_type" + t.integer "file_size" + t.datetime "processed_at" t.index ["account_id"], name: "index_captain_documents_on_account_id" t.index ["assistant_id", "external_link"], name: "index_captain_documents_on_assistant_id_and_external_link", unique: true t.index ["assistant_id"], name: "index_captain_documents_on_assistant_id" + t.index ["content_type"], name: "index_captain_documents_on_content_type" + t.index ["document_type"], name: "index_captain_documents_on_document_type" + t.index ["source_type"], name: "index_captain_documents_on_source_type" t.index ["status"], name: "index_captain_documents_on_status" end @@ -443,6 +451,18 @@ ActiveRecord::Schema[7.1].define(version: 2025_05_23_031839) do t.index ["account_id", "profile_id"], name: "index_channel_twitter_profiles_on_account_id_and_profile_id", unique: true end + create_table "channel_voice", force: :cascade do |t| + t.string "phone_number", null: false + t.string "provider", default: "twilio", null: false + t.jsonb "provider_config", null: false + t.integer "account_id", null: false + t.jsonb "additional_attributes", default: {} + t.datetime "created_at", null: false + t.datetime "updated_at", null: false + t.index ["account_id"], name: "index_channel_voice_on_account_id" + t.index ["phone_number"], name: "index_channel_voice_on_phone_number", unique: true + end + create_table "channel_web_widgets", id: :serial, force: :cascade do |t| t.string "website_url" t.integer "account_id" @@ -813,6 +833,7 @@ ActiveRecord::Schema[7.1].define(version: 2025_05_23_031839) do t.text "processed_message_content" t.jsonb "sentiment", default: {} t.index "((additional_attributes -> 'campaign_id'::text))", name: "index_messages_on_additional_attributes_campaign_id", using: :gin + t.index ["account_id", "content_type", "created_at"], name: "idx_messages_account_content_created" t.index ["account_id", "created_at", "message_type"], name: "index_messages_on_account_created_type" t.index ["account_id", "inbox_id"], name: "index_messages_on_account_id_and_inbox_id" t.index ["account_id"], name: "index_messages_on_account_id" @@ -907,7 +928,7 @@ ActiveRecord::Schema[7.1].define(version: 2025_05_23_031839) do t.text "header_text" t.datetime "created_at", null: false t.datetime "updated_at", null: false - t.jsonb "config", default: {"allowed_locales"=>["en"]} + t.jsonb "config", default: {"allowed_locales" => ["en"]} t.boolean "archived", default: false t.bigint "channel_web_widget_id" t.index ["channel_web_widget_id"], name: "index_portals_on_channel_web_widget_id" diff --git a/enterprise/app/controllers/api/v1/accounts/captain/documents_controller.rb b/enterprise/app/controllers/api/v1/accounts/captain/documents_controller.rb index 3c7b86607..e5b43d712 100644 --- a/enterprise/app/controllers/api/v1/accounts/captain/documents_controller.rb +++ b/enterprise/app/controllers/api/v1/accounts/captain/documents_controller.rb @@ -6,7 +6,12 @@ class Api::V1::Accounts::Captain::DocumentsController < Api::V1::Accounts::BaseC before_action :set_documents, except: [:create] before_action :set_document, only: [:show, :destroy] before_action :set_assistant, only: [:create, :upload_pdf] + RESULTS_PER_PAGE = 25 + # Fixed PDF size limit + MAX_PDF_SIZE = 25.megabytes + ALLOWED_PDF_CONTENT_TYPES = ['application/pdf'].freeze + PDF_MAGIC_NUMBERS = ['%PDF'].freeze def index base_query = @documents @@ -28,32 +33,16 @@ class Api::V1::Accounts::Captain::DocumentsController < Api::V1::Accounts::BaseC end def upload_pdf - return render_could_not_create_error('Missing Assistant') if @assistant.nil? - return render_could_not_create_error('No PDF file provided') unless pdf_file_present? - - # Validate PDF file - return render_could_not_create_error('Invalid file type. Only PDF files are allowed.') unless valid_pdf_file? - return render_could_not_create_error('File size too large. Maximum size is 10MB.') unless valid_pdf_size? - - # Upload PDF to storage and get URL - blob = create_pdf_blob - pdf_url = Rails.application.routes.url_helpers.rails_blob_url(blob, host: ENV.fetch('FRONTEND_URL', 'http://localhost:3000')) - - # Create document with PDF URL - @document = @assistant.documents.build( - name: pdf_params[:pdf_document].original_filename.gsub('.pdf', ''), - external_link: pdf_url - ) - @document.save! - - render json: { - document: @document.as_json(only: [:id, :name, :status, :created_at]), - message: 'PDF uploaded successfully. Processing will begin shortly.' - } + validation_errors = validate_upload_prerequisites + return render_could_not_create_error(validation_errors) if validation_errors + + process_pdf_upload rescue Captain::Document::LimitExceededError => e - render_could_not_create_error(e.message) + handle_limit_exceeded_error(e) + rescue ActiveStorage::FileNotFoundError => e + handle_file_not_found_error(e) rescue StandardError => e - render_could_not_create_error("PDF upload failed: #{e.message}") + handle_general_upload_error(e) end def destroy @@ -96,23 +85,99 @@ class Api::V1::Accounts::Captain::DocumentsController < Api::V1::Accounts::BaseC pdf_params[:pdf_document].present? end - def valid_pdf_file? - return false unless pdf_params[:pdf_document].respond_to?(:content_type) - - pdf_params[:pdf_document].content_type == 'application/pdf' - end + def validate_pdf_file + file = pdf_params[:pdf_document] + return { valid: false, error: 'Invalid file object' } unless file.respond_to?(:content_type) && file.respond_to?(:size) - def valid_pdf_size? - return false unless pdf_params[:pdf_document].respond_to?(:size) - - pdf_params[:pdf_document].size <= 10.megabytes + return { valid: false, error: 'Invalid file type. Only PDF files are allowed.' } unless ALLOWED_PDF_CONTENT_TYPES.include?(file.content_type) + + return { valid: false, error: "File size too large. Maximum size is #{MAX_PDF_SIZE / 1.megabyte}MB." } if file.size > MAX_PDF_SIZE + + { valid: true } end def create_pdf_blob + file = pdf_params[:pdf_document] + ActiveStorage::Blob.create_and_upload!( - io: pdf_params[:pdf_document].tempfile, - filename: pdf_params[:pdf_document].original_filename, - content_type: pdf_params[:pdf_document].content_type + io: file.tempfile, + filename: sanitize_filename(file.original_filename), + content_type: file.content_type, + metadata: { + uploaded_by: Current.user&.id, + assistant_id: @assistant.id, + account_id: Current.account.id, + original_filename: file.original_filename + } + ) + rescue StandardError => e + Rails.logger.error "Failed to create PDF blob: #{e.message}" + raise ActiveStorage::FileNotFoundError, 'Failed to upload PDF file' + end + + def generate_pdf_url(blob) + Rails.application.routes.url_helpers.rails_blob_url( + blob, + host: ENV.fetch('FRONTEND_URL') { Rails.application.config.action_mailer.default_url_options[:host] } ) end + + def sanitize_filename(filename) + return 'document' if filename.blank? + + base_name = File.basename(filename, '.pdf') + sanitized = base_name.gsub(/[^\w\s.-]/, '').strip.squeeze(' ') + (sanitized.presence || 'document') + end + + def process_pdf_upload + ActiveRecord::Base.transaction do + blob = create_pdf_blob + pdf_url = generate_pdf_url(blob) + + @document = @assistant.documents.build( + name: sanitize_filename(pdf_params[:pdf_document].original_filename), + external_link: pdf_url, + source_type: 'pdf_upload', + content_type: pdf_params[:pdf_document].content_type, + file_size: pdf_params[:pdf_document].size + ) + @document.save! + + log_pdf_upload_success + + # Use the same response structure as create action for consistency + render :create + end + end + + def log_pdf_upload_success + Rails.logger.info "PDF uploaded successfully - Document ID: #{@document.id}, Assistant ID: #{@assistant.id}, Account ID: #{Current.account.id}" + end + + def validate_upload_prerequisites + return 'Missing Assistant' if @assistant.nil? + return 'No PDF file provided' unless pdf_file_present? + + validation_result = validate_pdf_file + return validation_result[:error] unless validation_result[:valid] + + nil + end + + def handle_limit_exceeded_error(error) + Rails.logger.warn "Document limit exceeded for assistant #{@assistant.id}: #{error.message}" + render_could_not_create_error(error.message) + end + + def handle_file_not_found_error(error) + Rails.logger.error "PDF file not found during upload: #{error.message}" + render_could_not_create_error('PDF file could not be processed. Please try again.') + end + + def handle_general_upload_error(error) + Rails.logger.error "PDF upload failed for assistant #{@assistant&.id}: #{error.message}" + Rails.logger.error error.backtrace.join("\n") + render_could_not_create_error('PDF upload failed. Please try again.') + end end diff --git a/enterprise/app/jobs/captain/documents/crawl_job.rb b/enterprise/app/jobs/captain/documents/crawl_job.rb index 86e994aa9..00e8cbe34 100644 --- a/enterprise/app/jobs/captain/documents/crawl_job.rb +++ b/enterprise/app/jobs/captain/documents/crawl_job.rb @@ -2,59 +2,56 @@ class Captain::Documents::CrawlJob < ApplicationJob queue_as :low def perform(document) - if InstallationConfig.find_by(name: 'CAPTAIN_FIRECRAWL_API_KEY')&.value.present? + if pdf_document?(document) + perform_pdf_extraction(document) + elsif InstallationConfig.find_by(name: 'CAPTAIN_FIRECRAWL_API_KEY')&.value.present? perform_firecrawl_crawl(document) else - if pdf_document?(document) - perform_pdf_extraction(document) - else - perform_simple_crawl(document) - end + perform_simple_crawl(document) end end + def pdf_document?(document) + return false if document.nil? + + pdf_by_metadata?(document) || pdf_by_url?(document) + end + private include Captain::FirecrawlHelper - def pdf_document?(document) - # Check if the document source is a PDF - document.external_link&.downcase&.end_with?('.pdf') || - document.content_type&.include?('application/pdf') || - document.source_type == 'pdf_upload' + def pdf_by_metadata?(document) + return true if document.respond_to?(:source_type) && document.source_type == 'pdf_upload' + return true if document.respond_to?(:content_type) && document.content_type&.include?('application/pdf') + + false + rescue StandardError + false + end + + def pdf_by_url?(document) + return false if document.external_link.blank? + + url = document.external_link.downcase + url.end_with?('.pdf') || + url.include?('/rails/active_storage/blobs/') || + (url.include?('blob') && url.include?('pdf')) end def perform_pdf_extraction(document) - begin - # Mark document as in progress - document.update(status: 'in_progress') + document.update(status: 'in_progress') - pdf_extraction_service = Captain::Tools::PdfExtractionService.new(document.external_link) - result = pdf_extraction_service.perform + pdf_extraction_service = Captain::Tools::PdfExtractionService.new(document.external_link) + result = pdf_extraction_service.perform - if result[:success] && result[:content].present? - # Process content in chunks similar to web crawling - result[:content].each do |content_chunk| - Captain::Tools::PdfExtractionParserJob.perform_later( - assistant_id: document.assistant_id, - pdf_content: content_chunk, - document_id: document.id - ) - end - - # Update document status to indicate processing has started - document.update(status: 'in_progress', processed_at: nil) - else - # Handle extraction failure - mark as available but log error - error_message = result[:errors]&.join(', ') || 'Failed to extract text from PDF' - document.update(status: 'available') - Rails.logger.error "PDF extraction failed for document #{document.id}: #{error_message}" - end - rescue StandardError => e - # Handle errors gracefully - mark as available but log error - Rails.logger.error "PDF extraction failed for document #{document.id}: #{e.message}" - document.update(status: 'available') + if result[:success] && result[:content].present? + process_pdf_content_chunks(document, result[:content]) + else + handle_pdf_extraction_failure(document, result) end + rescue StandardError => e + handle_pdf_extraction_error(document, e) end def perform_simple_crawl(document) @@ -92,4 +89,42 @@ class Captain::Documents::CrawlJob < ApplicationJob "#{webhook_url}?assistant_id=#{document.assistant_id}&token=#{generate_firecrawl_token(document.assistant_id, document.account_id)}" end + + def process_pdf_content_chunks(document, content_chunks) + Rails.logger.info "PDF extraction successful for document #{document.id}: #{content_chunks.length} chunks will be processed" + + content_chunks.each_with_index do |content_chunk, index| + log_chunk_queueing(document, content_chunk, index, content_chunks.length) + queue_pdf_chunk_job(document, content_chunk) + end + + document.update(status: 'in_progress', processed_at: nil) + Rails.logger.info "All #{content_chunks.length} chunks queued for processing for document #{document.id}" + end + + def log_chunk_queueing(document, content_chunk, index, total_chunks) + page_num = content_chunk[:page_number] + content_length = content_chunk[:content].length + Rails.logger.info "Queueing chunk #{index + 1}/#{total_chunks} for document #{document.id} " \ + "(Page #{page_num}, #{content_length} chars)" + end + + def queue_pdf_chunk_job(document, content_chunk) + Captain::Tools::PdfExtractionParserJob.perform_later( + assistant_id: document.assistant_id, + pdf_content: content_chunk, + document_id: document.id + ) + end + + def handle_pdf_extraction_failure(document, result) + error_message = result[:errors]&.join(', ') || 'Failed to extract text from PDF' + Rails.logger.error "PDF extraction failed for document #{document.id}: #{error_message}" + document.update(status: 'available') + end + + def handle_pdf_extraction_error(document, error) + Rails.logger.error "PDF extraction failed for document #{document.id}: #{error.message}" + document.update(status: 'available') + end end diff --git a/enterprise/app/jobs/captain/documents/response_builder_job.rb b/enterprise/app/jobs/captain/documents/response_builder_job.rb index 47b22112d..b7d0d5069 100644 --- a/enterprise/app/jobs/captain/documents/response_builder_job.rb +++ b/enterprise/app/jobs/captain/documents/response_builder_job.rb @@ -1,13 +1,16 @@ class Captain::Documents::ResponseBuilderJob < ApplicationJob queue_as :low - def perform(document) - # Skip processing if document has no content (e.g., PDF parent documents) - return if document.content.blank? + def perform(document, full_content = nil) + # Use full content for FAQ generation if provided (for PDFs), otherwise use document.content + content_for_faqs = full_content || document.content + + # Skip processing if no content available + return if content_for_faqs.blank? reset_previous_responses(document) - faqs = Captain::Llm::FaqGeneratorService.new(document.content).generate + faqs = Captain::Llm::FaqGeneratorService.new(content_for_faqs).generate faqs.each do |faq| create_response(faq, document) end diff --git a/enterprise/app/jobs/captain/tools/pdf_extraction_parser_job.rb b/enterprise/app/jobs/captain/tools/pdf_extraction_parser_job.rb index 623c63779..3a1f1d0d2 100644 --- a/enterprise/app/jobs/captain/tools/pdf_extraction_parser_job.rb +++ b/enterprise/app/jobs/captain/tools/pdf_extraction_parser_job.rb @@ -1,77 +1,134 @@ class Captain::Tools::PdfExtractionParserJob < ApplicationJob queue_as :low + retry_on StandardError, wait: :exponentially_longer, attempts: 3 + + # Redis and content management settings + REDIS_KEY_TTL = 7200 # 2 hours - increased for larger PDFs + DB_STORAGE_LIMIT = 50_000 # Store only first 50k chars in DB for search/preview def perform(assistant_id:, pdf_content:, document_id: nil) - assistant = Captain::Assistant.find(assistant_id) - account = assistant.account + validate_inputs!(assistant_id, pdf_content, document_id) - if limit_exceeded?(account) - Rails.logger.info("Document limit exceeded for assistant #{assistant_id}") - return - end + assistant = load_assistant(assistant_id) + return if processing_should_skip?(assistant) - return unless document_id.present? - - # Find the original document - document = Captain::Document.find_by(id: document_id) + document = load_document(document_id) return unless document - # Extract content from PDF content hash - content = pdf_content[:content] || '' - page_number = pdf_content[:page_number] || 1 - chunk_index = pdf_content[:chunk_index] || 1 - total_chunks = pdf_content[:total_chunks] || 1 - - # Append content to the original document - append_content_to_document(document, content, page_number, chunk_index, total_chunks) + content_data = extract_content_data(pdf_content) + log_chunk_processing(document_id, content_data) + append_content_to_document(document, content_data) + rescue ActiveRecord::RecordNotFound => e + handle_record_not_found_error(e, document_id) rescue StandardError => e - Rails.logger.error "Failed to parse PDF content for assistant #{assistant_id}: #{e.message}" - - # Mark the document as available even if there's an error, so it doesn't get stuck - document&.update(status: 'available') if document_id.present? - - raise "Failed to parse PDF data: #{e.message}" + handle_processing_error(e, assistant_id, document_id) end private + def validate_inputs!(assistant_id, pdf_content, document_id) + raise ArgumentError, 'Assistant ID is required' if assistant_id.blank? + raise ArgumentError, 'PDF content is required' if pdf_content.blank? + raise ArgumentError, 'Document ID is required' if document_id.blank? + end + + def extract_content_data(pdf_content) + { + content: pdf_content[:content] || '', + page_number: pdf_content[:page_number] || 1, + chunk_index: pdf_content[:chunk_index] || 1, + total_chunks: pdf_content[:total_chunks] || 1 + } + end + + def mark_document_as_available(document_id) + return if document_id.blank? + + Captain::Document.find_by(id: document_id)&.update(status: 'available') + rescue StandardError => e + Rails.logger.error "Failed to mark document #{document_id} as available: #{e.message}" + end + def limit_exceeded?(account) limits = account.usage_limits.dig(:captain, :documents) return false unless limits - + limits[:current_available].to_i <= 0 end - def append_content_to_document(document, content, page_number, chunk_index, total_chunks) - # Use Redis to coordinate content aggregation across multiple jobs - redis_key = "pdf_content_#{document.id}" - - # Store this chunk's content in Redis with metadata - chunk_data = { - content: content, - page_number: page_number, - chunk_index: chunk_index, - total_chunks: total_chunks, + def build_redis_key(document_id) + "pdf_content_#{document_id}" + end + + def build_chunk_data(content_data) + { + content: content_data[:content], + page_number: content_data[:page_number], + chunk_index: content_data[:chunk_index], + total_chunks: content_data[:total_chunks], processed_at: Time.current.to_i } - - # Add chunk to Redis hash using Alfred - $alfred.with do |conn| - conn.hset(redis_key, "#{page_number}_#{chunk_index}", chunk_data.to_json) - conn.expire(redis_key, 3600) # Expire after 1 hour - end - - # Check if all chunks are processed - all_chunks = $alfred.with { |conn| conn.hgetall(redis_key) } - total_expected_chunks = calculate_total_expected_chunks(all_chunks) - - if all_chunks.size >= total_expected_chunks - # All chunks processed, aggregate content - aggregate_and_update_document(document, all_chunks, redis_key) + end + + def store_chunk_in_redis(redis_key, content_data, chunk_data) + chunk_key = "#{content_data[:page_number]}_#{content_data[:chunk_index]}" + + redis_connection.with do |conn| + conn.hset(redis_key, chunk_key, chunk_data.to_json) + conn.expire(redis_key, REDIS_KEY_TTL) end end - + + def retrieve_all_chunks(redis_key) + redis_connection.with { |conn| conn.hgetall(redis_key) } + end + + def ready_for_aggregation?(current_chunks, expected_chunks) + current_chunks >= expected_chunks + end + + def parse_and_sort_chunks(all_chunks) + parsed_chunks = all_chunks.map do |_key, chunk_json| + chunk_data = JSON.parse(chunk_json) + [chunk_data['page_number'], chunk_data['chunk_index'], chunk_data['content']] + end + + parsed_chunks.sort_by { |page, chunk_idx, _| [page, chunk_idx] } + rescue JSON::ParserError => e + Rails.logger.error "Failed to parse chunk JSON: #{e.message}" + raise 'Invalid chunk data format' + end + + def combine_chunks_content(sorted_chunks) + sorted_chunks.map(&:last).join("\n\n") + end + + def cleanup_redis_key(redis_key) + redis_connection.with { |conn| conn.del(redis_key) } + rescue Redis::BaseError => e + Rails.logger.warn "Failed to cleanup Redis key #{redis_key}: #{e.message}" + end + + def append_content_to_document(document, content_data) + redis_key = build_redis_key(document.id) + + chunk_data = build_chunk_data(content_data) + + store_chunk_in_redis(redis_key, content_data, chunk_data) + + # Process each chunk individually with response builder + Captain::Documents::ResponseBuilderJob.perform_later(document, content_data[:content]) + + all_chunks = retrieve_all_chunks(redis_key) + total_expected_chunks = calculate_total_expected_chunks(all_chunks) + + finalize_document_processing(document, all_chunks, redis_key) if ready_for_aggregation?(all_chunks.size, total_expected_chunks) + rescue Redis::BaseError => e + Rails.logger.error "Redis error during PDF content processing: #{e.message}" + raise 'Failed to process PDF content due to storage error' + end + def calculate_total_expected_chunks(all_chunks) # Calculate total expected chunks based on the chunks we have all_chunks.values.map do |chunk_json| @@ -79,26 +136,52 @@ class Captain::Tools::PdfExtractionParserJob < ApplicationJob chunk_data['total_chunks'] end.max || 1 end - - def aggregate_and_update_document(document, all_chunks, redis_key) - # Parse and sort chunks by page and chunk index - sorted_chunks = all_chunks.map do |key, chunk_json| + + def finalize_document_processing(document, all_chunks, redis_key) + sorted_chunks = parse_and_sort_chunks(all_chunks) + combined_content = combine_chunks_content(sorted_chunks) + + log_document_processing_summary(document, all_chunks, combined_content) + + update_document_with_content(document, combined_content) + cleanup_redis_key(redis_key) + + Rails.logger.info "PDF content finalized for document #{document.id}: #{combined_content.length} characters" + rescue ActiveRecord::RecordInvalid => e + handle_document_update_error(e, document, redis_key) + end + + def log_document_processing_summary(document, all_chunks, combined_content) + log_finalization_info(document, all_chunks, combined_content) + chunk_summary = all_chunks.map do |_key, chunk_json| chunk_data = JSON.parse(chunk_json) - [chunk_data['page_number'], chunk_data['chunk_index'], chunk_data['content']] - end.sort_by { |page, chunk_idx, _| [page, chunk_idx] } - - # Combine all content - combined_content = sorted_chunks.map(&:last).join("\n\n") - - # Update the document with combined content - document.update!( - content: combined_content[0..199_999], # Respect the 200k character limit - status: 'available' - ) - - # Clean up Redis key - $alfred.with { |conn| conn.del(redis_key) } - - Rails.logger.info "Successfully aggregated PDF content for document #{document.id}" + "Page #{chunk_data['page_number']} Chunk #{chunk_data['chunk_index']}" + end + Rails.logger.info "Document #{document.id} chunks processed: #{chunk_summary.join(', ')}" + end + + def update_document_with_content(document, combined_content) + db_content = prepare_content_for_storage(combined_content) + + ActiveRecord::Base.transaction do + document.update!( + content: db_content, + status: 'available', + processed_at: Time.current + ) + Rails.logger.info "Document #{document.id} content saved: #{db_content.length} characters" + end + end + + def prepare_content_for_storage(combined_content) + return combined_content if combined_content.length <= DB_STORAGE_LIMIT + + "#{combined_content[0, DB_STORAGE_LIMIT]}... [Content truncated for storage - full content processed by AI]" + end + + def handle_document_update_error(error, document, redis_key) + Rails.logger.error "Failed to update document #{document.id}: #{error.message}" + cleanup_redis_key(redis_key) + raise end end \ No newline at end of file diff --git a/enterprise/app/models/captain/document.rb b/enterprise/app/models/captain/document.rb index d2a02f5b5..d65bf7e5c 100644 --- a/enterprise/app/models/captain/document.rb +++ b/enterprise/app/models/captain/document.rb @@ -4,8 +4,14 @@ # # id :bigint not null, primary key # content :text +# content_type :string +# document_type :integer default(0), not null # external_link :string not null +# faq_data :json +# file_size :integer # name :string +# processed_at :datetime +# source_type :string default("url") # status :integer default("in_progress"), not null # created_at :datetime not null # updated_at :datetime not null @@ -17,6 +23,10 @@ # index_captain_documents_on_account_id (account_id) # index_captain_documents_on_assistant_id (assistant_id) # index_captain_documents_on_assistant_id_and_external_link (assistant_id,external_link) UNIQUE +# index_captain_documents_on_content_type (content_type) +# index_captain_documents_on_document_type (document_type) +# index_captain_documents_on_faq_data (faq_data) USING gin +# index_captain_documents_on_source_type (source_type) # index_captain_documents_on_status (status) # class Captain::Document < ApplicationRecord @@ -29,8 +39,10 @@ class Captain::Document < ApplicationRecord validates :external_link, presence: true validates :external_link, uniqueness: { scope: :assistant_id } - validates :content, length: { maximum: 200_000 } + validates :content, length: { maximum: 400_000 } + validates :source_type, inclusion: { in: %w[url pdf_upload] } before_validation :ensure_account_id + before_validation :set_default_source_type enum status: { in_progress: 0, @@ -57,6 +69,8 @@ class Captain::Document < ApplicationRecord def enqueue_response_builder_job return if status != 'available' + # Skip auto-enqueue for PDFs as they handle FAQ generation manually with full content + return if pdf_document? Captain::Documents::ResponseBuilderJob.perform_later(self) end @@ -73,4 +87,37 @@ class Captain::Document < ApplicationRecord limits = account.usage_limits[:captain][:documents] raise LimitExceededError, 'Document limit exceeded' unless limits[:current_available].positive? end + + def set_default_source_type + return if source_type.present? + + # Determine type without calling pdf_document? to avoid circular dependency + self.source_type = if content_type&.include?('application/pdf') || pdf_url_format? + 'pdf_upload' + else + 'url' + end + end + + # Public method used by CrawlJob to determine if document is a PDF + def pdf_document? + pdf_upload? || pdf_content_type? || pdf_url_format? + end + + def pdf_upload? + source_type == 'pdf_upload' + end + + def pdf_content_type? + content_type&.include?('application/pdf') + end + + def pdf_url_format? + return false if external_link.blank? + + url = external_link.downcase + url.end_with?('.pdf') || + url.include?('/rails/active_storage/blobs/') || + (url.include?('blob') && url.include?('pdf')) + end end diff --git a/enterprise/app/services/captain/tools/pdf_content_chunking_concern.rb b/enterprise/app/services/captain/tools/pdf_content_chunking_concern.rb new file mode 100644 index 000000000..ea5aac8d3 --- /dev/null +++ b/enterprise/app/services/captain/tools/pdf_content_chunking_concern.rb @@ -0,0 +1,89 @@ +module Captain::Tools::PdfContentChunkingConcern + extend ActiveSupport::Concern + + private + + def chunk_content(page_contents, max_chunk_size: self.class::MAX_CHUNK_SIZE) + return [] if page_contents.blank? + + all_chunks = [] + total_page_chunks = 0 + + # First pass: calculate total chunks across all pages + page_contents.each do |page_content| + page_chunks = split_content_into_chunks(page_content[:content], max_chunk_size) + total_page_chunks += page_chunks.length + end + + chunk_index = 0 + page_contents.each do |page_content| + page_chunks = split_content_into_chunks(page_content[:content], max_chunk_size) + + page_chunks.each do |chunk_content| + chunk_index += 1 + all_chunks << build_chunk(chunk_content, page_content[:page_number], chunk_index, total_page_chunks) + end + end + + all_chunks + end + + def split_content_into_chunks(content, max_size) + return [content] if content.length <= max_size + + paragraphs = content.split(/\n\s*\n/) + chunks = [] + current_chunk = '' + + paragraphs.each do |paragraph| + if paragraph.length > max_size + add_chunk_if_present(chunks, current_chunk) + chunks.concat(split_paragraph_into_chunks(paragraph, max_size)) + current_chunk = '' + elsif ("#{current_chunk}\n\n#{paragraph}").length > max_size + add_chunk_if_present(chunks, current_chunk) + current_chunk = paragraph + else + current_chunk = current_chunk.blank? ? paragraph : "#{current_chunk}\n\n#{paragraph}" + end + end + + add_chunk_if_present(chunks, current_chunk) + chunks.presence || [content] + end + + def split_paragraph_into_chunks(paragraph, max_size) + sentences = paragraph.split(/(?<=[.!?])\s+/) + chunks = [] + current_chunk = '' + + sentences.each do |sentence| + if sentence.length > max_size + add_chunk_if_present(chunks, current_chunk) + chunks << sentence[0, max_size] + current_chunk = '' + elsif ("#{current_chunk} #{sentence}").length > max_size + add_chunk_if_present(chunks, current_chunk) + current_chunk = sentence + else + current_chunk = current_chunk.blank? ? sentence : "#{current_chunk} #{sentence}" + end + end + + add_chunk_if_present(chunks, current_chunk) + chunks + end + + def add_chunk_if_present(chunks, chunk) + chunks << chunk.strip if chunk.present? + end + + def build_chunk(content, page_number, chunk_index, total_chunks) + { + content: content, + page_number: page_number, + chunk_index: chunk_index, + total_chunks: total_chunks + } + end +end \ No newline at end of file diff --git a/enterprise/app/services/captain/tools/pdf_extraction_service.rb b/enterprise/app/services/captain/tools/pdf_extraction_service.rb index a1bdfe458..d6516f488 100644 --- a/enterprise/app/services/captain/tools/pdf_extraction_service.rb +++ b/enterprise/app/services/captain/tools/pdf_extraction_service.rb @@ -2,35 +2,43 @@ require 'pdf-reader' class Captain::Tools::PdfExtractionService include ActiveModel::Validations + include Captain::Tools::PdfValidationConcern + include Captain::Tools::PdfContentChunkingConcern attr_reader :pdf_source, :errors + # Fixed PDF processing limits + MAX_PDF_SIZE = 25.megabytes + MAX_CHUNK_SIZE = 10_000 + DOWNLOAD_TIMEOUT = 60 # Increased for larger file downloads + def initialize(pdf_source) @pdf_source = pdf_source @errors = [] end def perform - return { success: false, errors: ['Invalid PDF source'] } if pdf_source.blank? + return failure_response(['Invalid PDF source']) if pdf_source.blank? - begin - validate_pdf_source - content = extract_text - chunked_content = chunk_content(content) - { success: true, content: chunked_content } - rescue PDF::Reader::MalformedPDFError => e - Rails.logger.error "Malformed PDF: #{e.message}" - { success: false, errors: ['Invalid PDF format'] } - rescue StandardError => e - Rails.logger.error "PDF extraction error: #{e.message}" - { success: false, errors: [e.message] } - end + validation_result = validate_pdf_source + return validation_result unless validation_result[:success] + + process_pdf_extraction + rescue PDF::Reader::MalformedPDFError => e + handle_malformed_pdf_error(e) + rescue PDF::Reader::UnsupportedFeatureError => e + handle_unsupported_feature_error(e) + rescue Down::TimeoutError => e + handle_timeout_error(e) + rescue StandardError => e + handle_general_error(e) end def extract_text - if pdf_source.is_a?(String) && pdf_source.start_with?('http') - extract_from_url - elsif pdf_source.respond_to?(:tempfile) + case determine_source_type + when :url + active_storage_blob_url? ? extract_from_active_storage_blob : extract_from_url + when :uploaded_file extract_from_uploaded_file else extract_from_file_path @@ -39,36 +47,28 @@ class Captain::Tools::PdfExtractionService private - def validate_pdf_source - if pdf_source.is_a?(String) && pdf_source.start_with?('http') - validate_url - elsif pdf_source.respond_to?(:tempfile) - validate_uploaded_file - else - validate_file_path - end - end + def determine_source_type + return :url if pdf_source.is_a?(String) && pdf_source.start_with?('http') + return :uploaded_file if pdf_source.respond_to?(:tempfile) - def validate_url - uri = URI.parse(pdf_source) - raise StandardError, 'Invalid URL' unless uri.is_a?(URI::HTTP) || uri.is_a?(URI::HTTPS) - end - - def validate_uploaded_file - raise StandardError, 'File too large' if pdf_source.size > 10.megabytes - raise StandardError, 'Invalid file type' unless pdf_source.content_type == 'application/pdf' - end - - def validate_file_path - raise StandardError, 'File does not exist' unless File.exist?(pdf_source) - raise StandardError, 'File too large' if File.size(pdf_source) > 10.megabytes + :file_path end def extract_from_url - temp_file = Down.download(pdf_source) - result = extract_from_file(temp_file.path) - temp_file.close - temp_file.unlink + temp_file = Down.download( + pdf_source, + max_size: MAX_PDF_SIZE, + open_timeout: DOWNLOAD_TIMEOUT, + read_timeout: DOWNLOAD_TIMEOUT + ) + + begin + result = extract_from_file(temp_file.path) + ensure + temp_file.close + temp_file.unlink + end + result end @@ -80,24 +80,70 @@ class Captain::Tools::PdfExtractionService extract_from_file(pdf_source) end + def active_storage_blob_url? + pdf_source.include?('/rails/active_storage/blobs/') || + (pdf_source.include?('blob') && pdf_source.include?('pdf')) + end + + def extract_from_active_storage_blob + blob = find_blob_from_url + return extract_blob_content(blob) if blob + + Rails.logger.warn 'ActiveStorage blob not found, falling back to URL download' + extract_from_url + end + + def extract_blob_content(blob) + blob.open { |file| extract_from_file(file.path) } + end + + def find_blob_from_url + blob_key = extract_blob_key_from_url + return nil if blob_key.blank? + + ActiveStorage::Blob.find_by(key: blob_key) + rescue StandardError => e + Rails.logger.error "Error finding blob with key '#{blob_key}': #{e.message}" + nil + end + + def extract_blob_key_from_url + blob_key = if pdf_source.include?('/rails/active_storage/blobs/') + extract_key_from_rails_path + elsif pdf_source.include?('/blobs/') + extract_key_from_blob_path + end + + blob_key&.split('?')&.first # Remove query parameters + end + + def extract_key_from_rails_path + parts = pdf_source.split('/rails/active_storage/blobs/').last.split('/') + parts.length > 1 && parts[0] != 'redirect' ? parts[0] : parts[1] + end + + def extract_key_from_blob_path + pdf_source.split('/blobs/').last.split('/').first + end + def extract_from_file(file_path) text_content = [] - + PDF::Reader.open(file_path) do |reader| reader.pages.each_with_index do |page, index| - begin - page_text = page.text - next if page_text.blank? - - cleaned_text = clean_text(page_text) + page_text = page.text + next if page_text.blank? + + cleaned_text = clean_text(page_text) + if cleaned_text.present? text_content << { page_number: index + 1, content: cleaned_text - } if cleaned_text.present? - rescue StandardError => e - Rails.logger.warn "Failed to extract text from page #{index + 1}: #{e.message}" - next + } end + rescue StandardError => e + Rails.logger.warn "Failed to extract text from page #{index + 1}: #{e.message}" + next end end @@ -106,90 +152,74 @@ class Captain::Tools::PdfExtractionService def clean_text(text) # Remove form feeds and normalize whitespace - cleaned = text.gsub(/\f/, "\n") - .gsub(/\r\n/, "\n") - .gsub(/\r/, "\n") + cleaned = text.tr("\f", "\n") + .gsub("\r\n", "\n") + .tr("\r", "\n") .gsub(/\s+/, ' ') .gsub(/\n\s*\n\s*\n+/, "\n\n") .strip # Remove common PDF artifacts cleaned = cleaned.gsub(/^\d+\s*$/, '') # Remove standalone page numbers - .gsub(/^[\s\-_=]+$/, '') # Remove separator lines - .strip + .gsub(/^[\s\-_=]+$/, '') # Remove separator lines + .strip - cleaned.present? ? cleaned : nil + (cleaned.presence) end - def chunk_content(page_contents, max_chunk_size: 2000) - return [] if page_contents.blank? - - chunks = [] - - page_contents.each do |page_data| - content = page_data[:content] - page_number = page_data[:page_number] - - # If content is small enough, keep as single chunk - if content.length <= max_chunk_size - chunks << { - content: content, - page_number: page_number, - chunk_index: 1, - total_chunks: 1 - } - else - # Split large content into smaller chunks - page_chunks = split_content_into_chunks(content, max_chunk_size) - page_chunks.each_with_index do |chunk_content, index| - chunks << { - content: chunk_content, - page_number: page_number, - chunk_index: index + 1, - total_chunks: page_chunks.length - } - end - end + def pdf_source_type + case pdf_source + when String + pdf_source.start_with?('http') ? 'URL' : 'file_path' + else + 'uploaded_file' end - - chunks end - def split_content_into_chunks(content, max_size) - # Split by paragraphs first - paragraphs = content.split(/\n\s*\n/) - chunks = [] - current_chunk = "" + def log_extraction_success(chunked_content) + total_chars = chunked_content.sum { |chunk| chunk[:content].length } + Rails.logger.info "PDF extraction completed: #{chunked_content.length} chunks, #{total_chars} characters" + Rails.logger.info "PDF chunks breakdown: #{chunked_content.map do |chunk| + "Page #{chunk[:page_number]} (#{chunk[:content].length} chars)" + end.join(', ')}" + end - paragraphs.each do |paragraph| - # If adding this paragraph would exceed the limit - if (current_chunk + "\n\n" + paragraph).length > max_size - # Save current chunk if it has content - chunks << current_chunk.strip if current_chunk.present? - - # If single paragraph is too large, split by sentences - if paragraph.length > max_size - sentences = paragraph.split(/(?<=[.!?])\s+/) - current_chunk = "" - - sentences.each do |sentence| - if (current_chunk + " " + sentence).length > max_size - chunks << current_chunk.strip if current_chunk.present? - current_chunk = sentence - else - current_chunk += (current_chunk.present? ? " " : "") + sentence - end - end - else - current_chunk = paragraph - end - else - current_chunk += (current_chunk.present? ? "\n\n" : "") + paragraph - end - end + def process_pdf_extraction + content = extract_text + return failure_response(['No text content found in PDF']) if content.blank? - # Add the last chunk - chunks << current_chunk.strip if current_chunk.present? - chunks + chunked_content = chunk_content(content) + log_extraction_success(chunked_content) + + success_response(chunked_content) + end + + def success_response(content) + { success: true, content: content } + end + + def failure_response(errors) + { success: false, errors: errors } + end + + def handle_malformed_pdf_error(error) + Rails.logger.error "Malformed PDF (#{pdf_source_type}): #{error.message}" + failure_response(['Invalid or corrupted PDF format']) + end + + def handle_unsupported_feature_error(error) + Rails.logger.error "Unsupported PDF feature (#{pdf_source_type}): #{error.message}" + failure_response(['PDF contains unsupported features']) + end + + def handle_timeout_error(error) + Rails.logger.error "PDF download timeout (#{pdf_source_type}): #{error.message}" + failure_response(['PDF download timed out']) + end + + def handle_general_error(error) + Rails.logger.error "PDF extraction error (#{pdf_source_type}): #{error.message}" + Rails.logger.error error.backtrace.join("\n") + failure_response(['Failed to process PDF']) end end \ No newline at end of file diff --git a/enterprise/app/services/captain/tools/pdf_validation_concern.rb b/enterprise/app/services/captain/tools/pdf_validation_concern.rb new file mode 100644 index 000000000..b44b6bbc7 --- /dev/null +++ b/enterprise/app/services/captain/tools/pdf_validation_concern.rb @@ -0,0 +1,42 @@ +module Captain::Tools::PdfValidationConcern + extend ActiveSupport::Concern + + private + + def validate_pdf_source + case determine_source_type + when :url then validate_url + when :uploaded_file then validate_uploaded_file + else validate_file_path + end + + { success: true } + rescue StandardError => e + Rails.logger.error "PDF validation failed: #{e.message}" + { success: false, errors: [e.message] } + end + + def validate_url + uri = URI.parse(pdf_source) + raise StandardError, 'Invalid URL format' unless uri.is_a?(URI::HTTP) || uri.is_a?(URI::HTTPS) + raise StandardError, 'URL too long' if pdf_source.length > 2048 + raise StandardError, 'Invalid URL scheme' unless %w[http https].include?(uri.scheme) + rescue URI::InvalidURIError + raise StandardError, 'Malformed URL' + end + + def validate_uploaded_file + raise StandardError, 'File object is invalid' unless pdf_source.respond_to?(:size) && pdf_source.respond_to?(:content_type) + raise StandardError, "File too large (max #{self.class::MAX_PDF_SIZE / 1.megabyte}MB)" if pdf_source.size > self.class::MAX_PDF_SIZE + raise StandardError, 'Invalid file type' unless pdf_source.content_type == 'application/pdf' + raise StandardError, 'Empty file' if pdf_source.empty? + end + + def validate_file_path + raise StandardError, 'File path is blank' if pdf_source.blank? + raise StandardError, 'File does not exist' unless File.exist?(pdf_source) + raise StandardError, "File too large (max #{self.class::MAX_PDF_SIZE / 1.megabyte}MB)" if File.size(pdf_source) > self.class::MAX_PDF_SIZE + raise StandardError, 'Empty file' if File.empty?(pdf_source) + raise StandardError, 'File is not readable' unless File.readable?(pdf_source) + end +end \ No newline at end of file