Update main.py

.gitattributes

@@ -34,4 +34,3 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
-*.pdf filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -20,4 +20,3 @@ Desktop.ini
 *.bak
 *.tmp
 *.old
-assets/tmp/

Dockerfile

@@ -19,19 +19,14 @@ RUN pip install -r requirements.txt
 COPY --chown=user:user main.py .
 COPY --chown=user:user modules/ ./modules/
 COPY --chown=user:user config.cfg .
+COPY --chown=user:user product_classification.csv .
 
 # ---------- Copy examples folder explicitly ----------
 COPY --chown=user:user examples/ ./examples/
 
-# ---------- Copy assets folder ----------
-COPY --chown=user:user assets/ ./assets/
-
 # ---------- Ensure proper permissions for examples folder ----------
 RUN chmod -R 755 examples/
 
-# ---------- Ensure proper permissions for assets folder ----------
-RUN chmod -R 755 assets/
-
 # Switch to non-root user
 USER user
 

assets/label_title_mapping.csv

@@ -1,18 +0,0 @@
-language,category_100,category_mix,category_recycled
-german,100%,MIX,RECYCLED
-french,100%,MIXTE,RECYCLÉ
-spanish,100%,MIXTO,RECICLADO
-turkish,100%,KARMA,GERİKAZANILMIŞ
-romanian,100%,MIX,RECICLAT
-polish,100%,MIESZANY,RECYKLING
-portugese,100%,MISTO,RECICLADO
-norwegian,100%,MIX,RECYCLED
-swedish,100%,MIX,Återvunnet
-danish,100%,MIX,RECYCLED
-finnish,100%,MIX,KIERRÄTETTY
-chinese_simplified,100%,混合产品,再生产品
-chinese_traditional,100%,混合產品,再生產品
-russian,100%,МИКС,ВТОРИЧНЫЙ
-greek,100%,ΜΕΙΓΜΑ,ΑΝΑΚΥΚΛΩΜΕΝΟ
-japanese,100%,ミックス,リサイクル
-serbian,100%,МЕШАВИНА,РЕЦИКЛИРАНО

config.cfg

@@ -1,8 +1,10 @@
 [ocr]
-MODEL = o4-mini
+MODEL = gpt-4.1
+GUIDANCE_IMAGE = guidance.png
 
 [scope]
-MODEL = o4-mini
+TAXONOMY_CSV = product_classification.csv
+MODEL = gpt-5
 TEMPERATURE = 0
 
 [gradio]

main.py

@@ -1,32 +1,22 @@
 import os
 import logging
+import pandas as pd
+import openai
 import dotenv
 import gradio as gr
+import tempfile
+import json
 import configparser
 from pathlib import Path
+from typing import Optional, Dict, List, Tuple
+import time
 
 # Import from modules
+from modules.pdf_processing import get_crop_from_pdf
+from modules.llm_pipeline import extract_fsc_logo_and_taxonomy, build_taxonomy_block_from_dataframe
+from modules.fsc_api import get_fsc_products_for_license, filter_taxonomy_by_fsc_products
 from modules.image_processing import pdf_first_page_to_image
-from modules.pipeline import process_image_progressive
-
-# Load environment variables
-dotenv.load_dotenv()
-
-# ============================ Setup ============================================
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    datefmt='%H:%M:%S'
-)
-logger = logging.getLogger(__name__)
-
-# Load configuration
-config = configparser.ConfigParser()
-config.read('config.cfg')
-
-
-# ============================ Gradio setup ============================================
+from modules.ocr_extraction import extract_fsc_logo_data_single
 
 APP_DIR = Path(__file__).resolve().parent
 EXAMPLES_DIR = Path(os.environ.get("EXAMPLES_DIR", APP_DIR / "examples")).resolve()
@@ -47,25 +37,281 @@ def load_example_images():
     return files
 
 # when updating components programmatically in Gradio 4, prefer .update(...)
-def show_image_preview(file_path):
-    """Show preview of any image file (example or user upload)."""
-    if file_path:
-        # Handle PDF files by converting to image for thumbnail display
-        if file_path.lower().endswith('.pdf'):
+def select_example_image(example_path):
+    return example_path if example_path else None
+
+def show_example_thumbnail(example_path):
+    return example_path if example_path else None
+
+def select_and_show_example(example_path):
+    """Select example image and show thumbnail in one function."""
+    if example_path:
+        return example_path, example_path  # Return (image_input, thumbnail)
+    else:
+        return None, None
+
+# Load environment variables
+dotenv.load_dotenv()
+
+# ============================ Setup ============================================
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    datefmt='%H:%M:%S'
+)
+logger = logging.getLogger(__name__)
+
+# Load configuration
+config = configparser.ConfigParser()
+config.read('config.cfg')
+
+# Expect your API key in environment: OPENAI_API_KEY
+client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+# Model configuration
+OCR_MODEL = config.get('ocr', 'MODEL')
+SCOPE_MODEL = config.get('scope', 'MODEL')
+
+# File paths
+GUIDANCE_IMAGE = config.get('ocr', 'GUIDANCE_IMAGE')
+TAXONOMY_CSV = config.get('scope', 'TAXONOMY_CSV')
+
+# Load taxonomy once at startup
+try:
+    taxonomy_df = pd.read_csv(TAXONOMY_CSV)
+    logger.info(f"Loaded taxonomy with {len(taxonomy_df)} categories")
+except Exception as e:
+    logger.error(f"Failed to load taxonomy: {e}")
+    taxonomy_df = pd.DataFrame()
+
+def step1_extract_fsc_code(image_file) -> Tuple[str, str, str, str, str, str]:
+    """Step 1: Extract FSC license code using OCR."""
+    if image_file is None:
+        return "Please upload an image file.", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", ""
+    
+    try:
+        # Handle the uploaded file
+        if isinstance(image_file, str):
+            image_path = image_file
+        else:
+            image_path = image_file.name
+        
+        logger.info(f"Processing image: {image_path}")
+        
+        # Convert PDF to image if needed
+        cleanup_png = False
+        try:
+            if image_path.lower().endswith(".pdf"):
+                img_path = pdf_first_page_to_image(image_path)
+                cleanup_png = True
+                logger.info("Converted PDF to image")
+            elif image_path.lower().endswith((".png", ".jpg", ".jpeg", ".webp")):
+                img_path = image_path
+            else:
+                return "Unsupported file format. Please upload PDF, PNG, JPG, JPEG, or WEBP files.", "", "", "", "", ""
+            
+            # Step 1: Extract FSC license code using OCR extraction
+            logger.info("Step 1: Extracting FSC license code using OCR...")
+            
+            # Initialize token and cost tracking
+            total_tokens = {"input": 0, "output": 0}
+            total_cost = [0.0]
+            
             try:
-                # Convert PDF to image for thumbnail display
-                thumbnail_path = pdf_first_page_to_image(file_path, max_long_edge=400, fmt="PNG")
-                return file_path, thumbnail_path  # Return (original_file, thumbnail_image)
+                # Use the single-file OCR extraction function
+                ocr_data = extract_fsc_logo_data_single(
+                    img_path, 
+                    client, 
+                    model_name=OCR_MODEL,
+                    total_tokens=total_tokens,
+                    total_cost=total_cost
+                )
+                
+                # Extract FSC license code from OCR results
+                extracted_fsc_license_code = None
+                if ocr_data.get("fsc_data") and len(ocr_data["fsc_data"]) > 0:
+                    fsc_entry = ocr_data["fsc_data"][0]
+                    extracted_fsc_license_code = fsc_entry.get("fsc_license_code")
+                    if extracted_fsc_license_code:
+                        # Ensure it has the FSC- prefix
+                        if not extracted_fsc_license_code.startswith("FSC-"):
+                            extracted_fsc_license_code = f"FSC-{extracted_fsc_license_code}"
+                
+                logger.info(f"OCR extracted FSC license code: {extracted_fsc_license_code}")
+                
+                fsc_code_result = extracted_fsc_license_code or "Not found"
+                
+                return fsc_code_result, "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", ""
+                
             except Exception as e:
-                logger.error(f"Failed to create thumbnail for PDF {file_path}: {e}")
-                return file_path, None  # Return original file, no thumbnail
+                logger.error(f"OCR extraction failed: {e}")
+                return f"Error in OCR extraction: {str(e)}", "", "", "", "", ""
+            
+        finally:
+            # Clean up temporary PNG if created
+            if cleanup_png and 'img_path' in locals():
+                try:
+                    os.remove(img_path)
+                    logger.debug(f"Cleaned up temporary PNG: {img_path}")
+                except OSError:
+                    logger.warning(f"Failed to clean up temporary PNG: {img_path}")
+                    
+    except Exception as e:
+        logger.error(f"Error processing image: {e}")
+        return f"Error processing image: {str(e)}", "", "", "", "", ""
+
+def step2_fsc_lookup(image_file, fsc_code) -> Tuple[str, str, str, str, str, str]:
+    """Step 2: Lookup FSC license in API and get scope."""
+    if image_file is None:
+        return fsc_code, "Please upload an image file.", "", "", "", ""
+    
+    if not fsc_code or fsc_code == "Not found":
+        return fsc_code, "No FSC code found - skipping API lookup", "", "", "", ""
+    
+    try:
+        logger.info(f"Step 2: Looking up FSC license: {fsc_code}")
+        
+        fsc_products, license_status = get_fsc_products_for_license(fsc_code)
+        
+        return fsc_code, license_status or "Unknown", "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", ""
+        
+    except Exception as e:
+        logger.error(f"Error in FSC lookup: {e}")
+        return fsc_code, "Error", "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", ""
+
+def step3_analyze_with_filtered_taxonomy(image_file, fsc_code, fsc_products, license_status) -> Tuple[str, str, str, str, str, str]:
+    """Step 3: Analyze product with FSC-filtered taxonomy."""
+    if image_file is None:
+        return fsc_code, "Please upload an image file.", "", "", "", ""
+    
+    try:
+        # Handle the uploaded file
+        if isinstance(image_file, str):
+            image_path = image_file
         else:
-            # For image files, use the file directly for both
-            return file_path, file_path
-    else:
-        return None, None
+            image_path = image_file.name
+        
+        # Convert PDF to image if needed
+        cleanup_png = False
+        try:
+            if image_path.lower().endswith(".pdf"):
+                img_path = pdf_first_page_to_image(image_path)
+                cleanup_png = True
+            elif image_path.lower().endswith((".png", ".jpg", ".jpeg", ".webp")):
+                img_path = image_path
+            else:
+                return fsc_code, license_status or "Unknown", "Unsupported file format.", "", "", ""
+            
+            # Step 3: Filter taxonomy and analyze
+            logger.info("Step 3: Analyzing product with filtered taxonomy...")
+            
+            # Determine which taxonomy to use
+            if fsc_code and fsc_code != "Not found" and license_status == 'Valid' and fsc_products:
+                # Use filtered taxonomy based on FSC products
+                filtered_taxonomy = filter_taxonomy_by_fsc_products(taxonomy_df, fsc_products)
+                if not filtered_taxonomy.empty:
+                    taxonomy_block = build_taxonomy_block_from_dataframe(filtered_taxonomy, include_level3=True)
+                    taxonomy_source = "FSC-filtered"
+                else:
+                    taxonomy_block = build_taxonomy_block_from_dataframe(taxonomy_df, include_level3=True)
+                    taxonomy_source = "full (no FSC matches found)"
+            else:
+                # Use full taxonomy
+                taxonomy_block = build_taxonomy_block_from_dataframe(taxonomy_df, include_level3=True)
+                taxonomy_source = "full"
+            
+            logger.info(f"Using {taxonomy_source} taxonomy for analysis")
+            
+            # Initialize token and cost tracking
+            total_tokens = {"input": 0, "output": 0}
+            total_cost = [0.0]
+            
+            result = extract_fsc_logo_and_taxonomy(
+                img_path, 
+                taxonomy_block, 
+                original_filename=os.path.basename(image_path),
+                client=client, 
+                model_name=SCOPE_MODEL,
+                total_tokens=total_tokens, 
+                total_cost=total_cost
+            )
+            
+            # Format results
+            product_summary = result.get("product_summary", "No product summary available")
+            material = result.get("inferred_material", "No material information available")
+            
+            # Format taxonomy results
+            taxonomy_matches = result.get("taxonomy_matches", [])
+            has_match = result.get("has_taxonomy_match", False)
+            
+            taxonomy_results = ""
+            if has_match and taxonomy_matches:
+                for i, match in enumerate(taxonomy_matches[:3], 1):  # Show top 3 matches
+                    taxonomy_results += f"Match {i}:\n"
+                    if match.get("level1_code") and match.get("level1_name"):
+                        taxonomy_results += f"Level 1: {match['level1_code']} - {match['level1_name']}\n"
+                    if match.get("level2_code") and match.get("level2_name"):
+                        taxonomy_results += f"Level 2: {match['level2_code']} - {match['level2_name']}\n"
+                    if match.get("confidence"):
+                        taxonomy_results += f"Confidence: {match['confidence']:.2f}\n"
+                    if match.get("rationale"):
+                        taxonomy_results += f"Rationale: {match['rationale']}\n"
+                    taxonomy_results += "\n"
+            else:
+                taxonomy_results += "Taxonomy Classification: No matching categories found"
+            
+            cost_info = f"""
+Total Cost: ${total_cost[0]:.4f}
+Models Used: {OCR_MODEL} (OCR), {SCOPE_MODEL} (Analysis)
+Token Usage: {total_tokens['input']} input + {total_tokens['output']} output tokens"""
+            
+            return fsc_code, license_status or "Unknown", product_summary, material, taxonomy_results, cost_info
+            
+        finally:
+            # Clean up temporary PNG if created
+            if cleanup_png and 'img_path' in locals():
+                try:
+                    os.remove(img_path)
+                    logger.debug(f"Cleaned up temporary PNG: {img_path}")
+                except OSError:
+                    logger.warning(f"Failed to clean up temporary PNG: {img_path}")
+                    
+    except Exception as e:
+        logger.error(f"Error in step 3: {e}")
+        return fsc_code, license_status or "Unknown", f"Error in analysis: {str(e)}", "", "", ""
+
+def process_single_image_progressive(image_file):
+    """Process image with progressive updates showing each step."""
+    if image_file is None:
+        return "Please upload an image file.", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", ""
+    if image_file.lower().endswith(".pdf"):
+        image_file = get_crop_from_pdf(image_file)
+
+    if image_file is None:
+        return "Please upload an image file.", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", "⏳ Waiting for image...", ""
+
+    # Step 1: Extract FSC code
+    fsc_code, _, _, _, _, _ = step1_extract_fsc_code(image_file)
+    yield fsc_code, "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", "⏳ Step 2: Looking up FSC license...", ""
+
+    # Step 2: FSC lookup
+    fsc_code, license_status, _, _, _, _ = step2_fsc_lookup(image_file, fsc_code)
+    yield fsc_code, license_status, "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", "⏳ Step 3: Analyzing product...", ""
+
+    # Check if license status is valid before proceeding to step 3
+    if license_status != 'Valid':
+        # Skip step 3 if license is not valid
+        error_message = f"Analysis stopped: License status is '{license_status}', not 'Valid'. Cannot proceed with product analysis."
+        yield fsc_code, license_status, error_message, "", "Analysis skipped due to invalid license status", ""
+        return
 
-# ============================ Gradio UI ============================================
+    # Step 3: Analyze with filtered taxonomy
+    # We need to get the FSC products and license status from step 2
+    fsc_products, license_status = get_fsc_products_for_license(fsc_code) if fsc_code and fsc_code != "Not found" else ([], None)
+
+    fsc_code, license_status, product_summary, material, taxonomy_results, cost_info = step3_analyze_with_filtered_taxonomy(image_file, fsc_code, fsc_products, license_status)
+    yield fsc_code, license_status, product_summary, material, taxonomy_results, cost_info
 
 def create_gradio_interface():
     """Create and return the Gradio interface with progressive updates."""
@@ -85,7 +331,7 @@ def create_gradio_interface():
                 gr.Markdown("### Upload Your Own Image")
                 image_input = gr.File(
                     label="Upload Product Image",
-                    file_types=["image", ".pdf", ".PDF", ".Pdf", ".pDF"],
+                    file_types=["image", ".pdf"],
                     type="filepath"
                 )
 
@@ -126,23 +372,9 @@ def create_gradio_interface():
                 # Output components with formatted section headers
                 gr.Markdown("## Analysis Results")
 
-                gr.Markdown("### FSC Logo Data Extraction")
+                gr.Markdown("### FSC License Information")
                 fsc_license_output = gr.Textbox(
-                    label="FSC License Code",
-                    lines=1,
-                    interactive=False,
-                    container=True
-                )
-
-                label_title_output = gr.Textbox(
-                    label="Label Title",
-                    lines=1,
-                    interactive=False,
-                    container=True
-                )
-
-                product_type_output = gr.Textbox(
-                    label="Product Type",
+                    label="Extracted FSC License Code",
                     lines=1,
                     interactive=False,
                     container=True
@@ -155,7 +387,7 @@ def create_gradio_interface():
                     container=True
                 )
 
-                gr.Markdown("### Image Analysis")
+                gr.Markdown("### Product Analysis")
                 product_summary_output = gr.Textbox(
                     label="Product Description",
                     lines=3,
@@ -171,68 +403,14 @@ def create_gradio_interface():
                 )
 
                 taxonomy_output = gr.Textbox(
-                        label="Taxonomy Matches",
-                        lines=10,
-                        interactive=False,
-                        container=True
-                    )
-                image_text_output = gr.Textbox(
-                    label="Image Text",
-                    lines=15,
-                    interactive=False,
-                    container=True
-                )
-                fsc_text_strings_output = gr.Textbox(
-                    label="FSC Text Strings Identified (searching for FSC or Forest Stewardship Council)",
+                    label="Scope Check",
                     lines=10,
                     interactive=False,
                     container=True
                 )
-                gr.Markdown("### Scope Check")
-
-                overall_scope_check_output = gr.Textbox(
-                    label="Overall Compliance:",
-                    lines=2,
-                    interactive=False,
-                    container=True
-                )
-                materials_check_output = gr.Textbox(
-                    label="Product Type: Materials Check",
-                    lines=2,
-                    interactive=False,
-                    container=True
-                )
-                
-                taxonomy_check_output = gr.Textbox(
-                    label="Product Type: Taxonomy Check",
-                    lines=2,
-                    interactive=False,
-                    container=True
-                )
-
-                label_title_check_output = gr.Textbox(
-                    label="Label Title Check",
-                    lines=2,
-                    interactive=False,
-                    container=True
-                )
-
-                materials_check_rationale_output = gr.Textbox(
-                    label="Rationale: Materials Check",
-                    lines=5,
-                    interactive=False,
-                    container=True
-                )
-
-                taxonomy_check_rationale_output = gr.Textbox(
-                    label="Rationale: Taxonomy Check",
-                    lines=5,
-                    interactive=False,
-                    container=True
-                )
 
                 gr.Markdown("### Processing Information")
-                cost_info_output = gr.Textbox(
+                fsc_info_output = gr.Textbox(
                     label="Resources",
                     lines=3,
                     interactive=False,
@@ -243,22 +421,15 @@ def create_gradio_interface():
         if example_images:
             # Update thumbnail and load image when dropdown selection changes
             example_dropdown.change(
-                fn=show_image_preview,
+                fn=select_and_show_example,
                 inputs=[example_dropdown],
                 outputs=[image_input, example_thumbnail]
             )
 
-        # Update thumbnail when user uploads a file
-        image_input.change(
-            fn=show_image_preview,
-            inputs=[image_input],
-            outputs=[image_input, example_thumbnail]
-        )
-
         process_btn.click(
-            fn=process_image_progressive,
+            fn=process_single_image_progressive,
             inputs=[image_input],
-            outputs=[fsc_license_output, label_title_output, product_type_output, license_status_output, product_summary_output, material_output, taxonomy_output, image_text_output, fsc_text_strings_output, overall_scope_check_output, materials_check_output, taxonomy_check_output, label_title_check_output, materials_check_rationale_output, taxonomy_check_rationale_output, cost_info_output]
+            outputs=[fsc_license_output, license_status_output, product_summary_output, material_output, taxonomy_output, fsc_info_output]
         )
 
     return demo
@@ -267,6 +438,10 @@ def main():
     """Main function to launch the Gradio interface."""
     logger.info("Starting FSC Product Classification Tool with Gradio interface")
     
+    if taxonomy_df.empty:
+        logger.error("Failed to load taxonomy. Please check that product_classification.csv exists.")
+        return
+    
     if not os.getenv("OPENAI_API_KEY"):
         logger.error("OPENAI_API_KEY environment variable not set. Please set your OpenAI API key.")
         return

modules/{database_lookup.py → fsc_api.py}

@@ -1,5 +1,6 @@
 import os
 import json
+import time
 import logging
 from typing import List, Optional, Dict
 import pandas as pd
@@ -60,6 +61,20 @@ def call_fsc_api(license_code: str, api_base_url: str = None, api_key: str = Non
         logger.error(f"Error parsing JSON response for {license_code}: {e}")
         return None
 
+def extract_products_from_response(api_response: Dict) -> List[Dict]:
+    """
+    Extract the "Products" list from the API response.
+    
+    Args:
+        api_response (Dict): The API response dictionary
+        
+    Returns:
+        List[Dict]: List of product dictionaries, or empty list if not found
+    """
+    if not api_response:
+        return []
+    
+    return api_response.get("Products", [])
 
 def get_fsc_products_for_license(license_code: str, api_base_url: str = None, api_key: str = None) -> tuple[List[Dict], Optional[str]]:
     """
@@ -75,18 +90,18 @@ def get_fsc_products_for_license(license_code: str, api_base_url: str = None, ap
     """
     api_response = call_fsc_api(license_code, api_base_url, api_key)
     
-    # Check if API call failed
-    if api_response is None:
-        logger.warning(f"API call failed for {license_code} - license not found in FSC database")
-        return [], "Not Found"
+    # Check license status first
+    license_status = None
+    if api_response:
+        license_status = api_response.get("LicenseStatus")
+        logger.info(f"License status for {license_code}: {license_status}")
     
-    # Get license and certificate status
-    license_status = api_response.get("LicenseStatus")
-    certificate_status = api_response.get("CertificateStatus")
-    logger.info(f"Status for {license_code} - License: {license_status}; Certificate: {certificate_status}")
-    products = api_response.get("Products", [])
-
-    return products, license_status, certificate_status
+    products = extract_products_from_response(api_response)
+    
+    # Add a small delay to be respectful to the API
+    time.sleep(0.5)
+    
+    return products, license_status
 
 # ============================ Taxonomy Filtering ================================
 

modules/homography_functions.py

@@ -1,333 +0,0 @@
-#!/usr/bin/env python3
-"""
-Helper module for extracting logo_extraction.py - get logo from PDF pages using feature matching
-(SIFT/ORB) and homography estimation. 
-
-Key steps:
-- Compute keypoints/descriptors on page and template images.
-- Match descriptors with BFMatcher + Lowe's ratio test.
-- Estimate homography with RANSAC.
-- Return the homography matrix (M) if sufficient inliers are found.
-- Multiprocessing is used to run feature extraction with a timeout.
-
-Optimizations:
-- Removed expensive per-block multiprocessing overhead
-- Simplified feature extraction (no timeout handling per block)
-- Reduced default SIFT features for better performance
-- Maintained all validation logic
-- Added template inversion strategy
-- Dual matching approach (original + inverted)
-- Automatic selection of best match
-- Backward compatible with existing code
-"""
-
-import cv2
-import numpy as np
-from typing import Any, List, Optional, Tuple
-import logging
-
-logger = logging.getLogger(__name__)
-
-# ==== Global Constants ====
-MATCH_THRESHOLD: float = 0.65
-RANSAC_REPROJ_THRESHOLD: float = 5.0
-MIN_MATCHES: int = 8
-# ==== End Global Constants ====
-
-
-def serialize_keypoints(kps: List[cv2.KeyPoint]) -> List[Tuple]:
-    """Convert OpenCV KeyPoint objects into serializable format."""
-    return [
-        (kp.pt, kp.size, kp.angle, kp.response, kp.octave, kp.class_id)
-        for kp in kps
-    ]
-
-
-def deserialize_keypoints(serialized_kps: List[Tuple]) -> List[cv2.KeyPoint]:
-    """Reconstruct OpenCV KeyPoint objects from serialized tuples."""
-    return [
-        cv2.KeyPoint(
-            x=pt[0][0],
-            y=pt[0][1],
-            size=pt[1],
-            angle=pt[2],
-            response=pt[3],
-            octave=pt[4],
-            class_id=pt[5],
-        )
-        for pt in serialized_kps
-    ]
-
-
-def validate_homography(M: np.ndarray, template_shape: Tuple[int, int], 
-                       block_shape: Tuple[int, int]) -> bool:
-    """Validate that homography represents a reasonable transformation."""
-    if M is None:
-        return False
-    
-    h, w = template_shape
-    pts = np.float32([[0, 0], [0, h], [w, h], [w, 0]]).reshape(-1, 1, 2)
-    
-    try:
-        dst = cv2.perspectiveTransform(pts, M)
-    except:
-        logger.debug("Validation failed: perspectiveTransform error")
-        return False
-    
-    # Calculate transformed area
-    transformed_area = cv2.contourArea(dst)
-    original_area = w * h
-    
-    # Check area ratio
-    area_ratio = transformed_area / original_area
-    logger.debug(f"Area ratio: {area_ratio:.3f}")
-    if area_ratio < 0.01 or area_ratio > 100:
-        logger.debug(f"Area ratio {area_ratio:.3f} outside bounds")
-        return False
-    
-    # Check aspect ratio preservation
-    transformed_width = np.linalg.norm(dst[2] - dst[3])
-    transformed_height = np.linalg.norm(dst[1] - dst[0])
-    transformed_aspect = transformed_width / (transformed_height + 1e-6)
-    original_aspect = w / h
-    
-    aspect_ratio_change = transformed_aspect / original_aspect
-    logger.debug(f"Aspect ratio change: {aspect_ratio_change:.3f}")
-    if aspect_ratio_change < 0.5 or aspect_ratio_change > 2.0:
-        logger.debug(f"Aspect ratio change {aspect_ratio_change:.3f} outside bounds")
-        return False
-    
-    # Check for excessive skew
-    edge1 = dst[1] - dst[0]
-    edge2 = dst[3] - dst[0]
-    angle = np.abs(np.dot(edge1.flatten(), edge2.flatten()) / 
-                   (np.linalg.norm(edge1) * np.linalg.norm(edge2) + 1e-6))
-    logger.debug(f"Edge angle dot product: {angle:.3f}")
-    if angle > 0.5:
-        logger.debug(f"Excessive skew: {angle:.3f}")
-        return False
-    
-    logger.debug("Validation passed")
-    return True
-
-
-def try_match_with_template(
-    gray_page: np.ndarray,
-    gray_template: np.ndarray,
-    sift_or_orb: Any,
-    kp1: List[cv2.KeyPoint],
-    des1: np.ndarray,
-    inverted: bool = False
-) -> Tuple[Optional[np.ndarray], int, int]:
-    """
-    Try matching with a template (original or inverted).
-    
-    Returns
-    -------
-    Tuple[Optional[np.ndarray], int, int]
-        (Homography matrix, number of good matches, number of inliers)
-    """
-    # Extract features from page
-    try:
-        kp2, des2 = sift_or_orb.detectAndCompute(gray_page, None)
-    except Exception as e:
-        logger.debug(f"Feature extraction failed: {e}")
-        return None, 0, 0
-
-    if kp2 is None or des2 is None or len(des1) == 0 or len(des2) == 0:
-        return None, 0, 0
-
-    # Match descriptors
-    bf = cv2.BFMatcher()
-    matches = bf.knnMatch(des1, des2, k=2)
-    
-    # Apply Lowe's ratio test
-    good_matches = [
-        m for m_n in matches if len(m_n) == 2
-        for m, n in [m_n] if m.distance < MATCH_THRESHOLD * n.distance
-    ]
-
-    n_good = len(good_matches)
-    logger.debug(f"{'Inverted' if inverted else 'Original'} template: {n_good} good matches (need {MIN_MATCHES})")
-
-    if n_good < MIN_MATCHES:
-        return None, n_good, 0
-
-    # Prepare points and estimate homography
-    src_pts = np.float32([kp1[m.queryIdx].pt for m in good_matches]).reshape(-1, 1, 2)
-    dst_pts = np.float32([kp2[m.trainIdx].pt for m in good_matches]).reshape(-1, 1, 2)
-
-    M, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, RANSAC_REPROJ_THRESHOLD)
-    
-    if M is None:
-        logger.debug("Homography estimation failed")
-        return None, n_good, 0
-
-    # Validate inliers
-    inliers = np.sum(mask)
-    logger.debug(f"Inliers: {inliers}/{n_good}")
-    
-    if inliers < 10 or inliers / n_good < 0.6:
-        logger.debug("Insufficient inliers")
-        return None, n_good, inliers
-
-    # Validate geometric properties
-    if not validate_homography(M, gray_template.shape, gray_page.shape):
-        logger.debug("Homography failed geometric validation")
-        return None, n_good, inliers
-
-    return M, n_good, inliers
-
-
-def extract_logo_from_pdf_page(
-    page_cv: np.ndarray,
-    gray_template: np.ndarray,
-    sift_or_orb: Any,
-    kp1: List[cv2.KeyPoint],
-    des1: np.ndarray,
-) -> Optional[np.ndarray]:
-    """
-    Extract logo with polarity-invariant feature matching.
-    
-    This function tries both the original template and an inverted version,
-    then returns the homography from whichever gives better results.
-    
-    Parameters
-    ----------
-    page_cv : np.ndarray
-        Page image (BGR)
-    gray_template : np.ndarray
-        Template image (grayscale)
-    sift_or_orb : Any
-        Feature detector/descriptor
-    kp1 : List[cv2.KeyPoint]
-        Template keypoints (from original template)
-    des1 : np.ndarray
-        Template descriptors (from original template)
-        
-    Returns
-    -------
-    Optional[np.ndarray]
-        Homography matrix if logo found, None otherwise
-    """
-    gray_page = cv2.cvtColor(page_cv, cv2.COLOR_BGR2GRAY)
-
-    # Strategy 1: Try with original template
-    logger.debug("Trying original template...")
-    M_orig, n_good_orig, inliers_orig = try_match_with_template(
-        gray_page, gray_template, sift_or_orb, kp1, des1, inverted=False
-    )
-    
-    # Strategy 2: Try with inverted template (for white-on-dark logos)
-    logger.debug("Trying inverted template...")
-    gray_template_inv = 255 - gray_template
-    kp1_inv, des1_inv = sift_or_orb.detectAndCompute(gray_template_inv, None)
-    
-    M_inv, n_good_inv, inliers_inv = try_match_with_template(
-        gray_page, gray_template_inv, sift_or_orb, kp1_inv, des1_inv, inverted=True
-    )
-    
-    # Select the best match based on quality metrics
-    best_M = None
-    best_score = 0
-    
-    if M_orig is not None:
-        # Quality score: weighted combination of matches and inliers
-        score_orig = n_good_orig + (inliers_orig * 2)
-        if score_orig > best_score:
-            best_score = score_orig
-            best_M = M_orig
-            logger.info(f"Original template selected (score: {score_orig}, matches: {n_good_orig}, inliers: {inliers_orig})")
-    
-    if M_inv is not None:
-        # Quality score: weighted combination of matches and inliers
-        score_inv = n_good_inv + (inliers_inv * 2)
-        if score_inv > best_score:
-            best_score = score_inv
-            best_M = M_inv
-            logger.info(f"Inverted template selected (score: {score_inv}, matches: {n_good_inv}, inliers: {inliers_inv})")
-    
-    if best_M is not None:
-        logger.info("Logo detected with valid homography")
-        return best_M
-    else:
-        logger.debug("No valid match found with either template")
-        return None
-
-
-def extract_logo_from_pdf_page_dual_features(
-    page_cv: np.ndarray,
-    gray_template: np.ndarray,
-    sift_or_orb: Any,
-    kp1_orig: List[cv2.KeyPoint],
-    des1_orig: np.ndarray,
-    kp1_inv: List[cv2.KeyPoint],
-    des1_inv: np.ndarray,
-) -> Optional[np.ndarray]:
-    """
-    Extract logo with pre-computed inverted template features.
-    
-    This version is more efficient when processing multiple pages, as the
-    inverted template features are computed once and reused.
-    
-    Parameters
-    ----------
-    page_cv : np.ndarray
-        Page image (BGR)
-    gray_template : np.ndarray
-        Template image (grayscale, original)
-    sift_or_orb : Any
-        Feature detector/descriptor
-    kp1_orig : List[cv2.KeyPoint]
-        Template keypoints (from original template)
-    des1_orig : np.ndarray
-        Template descriptors (from original template)
-    kp1_inv : List[cv2.KeyPoint]
-        Template keypoints (from inverted template)
-    des1_inv : np.ndarray
-        Template descriptors (from inverted template)
-        
-    Returns
-    -------
-    Optional[np.ndarray]
-        Homography matrix if logo found, None otherwise
-    """
-    gray_page = cv2.cvtColor(page_cv, cv2.COLOR_BGR2GRAY)
-    gray_template_inv = 255 - gray_template
-
-    # Try with original template
-    logger.debug("Trying original template...")
-    M_orig, n_good_orig, inliers_orig = try_match_with_template(
-        gray_page, gray_template, sift_or_orb, kp1_orig, des1_orig, inverted=False
-    )
-    
-    # Try with inverted template
-    logger.debug("Trying inverted template...")
-    M_inv, n_good_inv, inliers_inv = try_match_with_template(
-        gray_page, gray_template_inv, sift_or_orb, kp1_inv, des1_inv, inverted=True
-    )
-    
-    # Select the best match
-    best_M = None
-    best_score = 0
-    
-    if M_orig is not None:
-        score_orig = n_good_orig + (inliers_orig * 2)
-        if score_orig > best_score:
-            best_score = score_orig
-            best_M = M_orig
-            logger.info(f"Original template selected (score: {score_orig})")
-    
-    if M_inv is not None:
-        score_inv = n_good_inv + (inliers_inv * 2)
-        if score_inv > best_score:
-            best_score = score_inv
-            best_M = M_inv
-            logger.info(f"Inverted template selected (score: {score_inv})")
-    
-    if best_M is not None:
-        logger.info("Logo detected with valid homography")
-        return best_M
-    else:
-        logger.debug("No valid match found with either template")
-        return None

modules/image_analysis.py

@@ -1,157 +0,0 @@
-import os
-import logging
-from typing import Dict
-import pandas as pd
-import openai
-import re
-
-# import from modules
-from modules.models import FSCExtractionAndTaxonomy
-from modules.prompts import prompt_image_analysis
-from modules.utils import get_token_costs
-from modules.image_processing import file_to_b64, pdf_multi_page_to_images
-
-# Configure logging
-logger = logging.getLogger(__name__)
-
-# ============================ Taxonomy Preparation ===================================
-def build_taxonomy_block_from_dataframe(taxonomy_df: pd.DataFrame, include_level3: bool = False) -> str:
-    """
-    Produces compact lines from a filtered taxonomy dataframe such as:
-      W12 Indoor furniture | W12.1 Cabinet
-      P5 Packaging and wrappings of paper | P5.1 Cardboard packaging | P5.2 Corrugated paper packaging
-    """
-    lines, seen = [], set()
-    
-    for _, row in taxonomy_df.iterrows():
-        # Handle NaN values properly
-        main_category = str(row.get("Main Category") or "").strip() if pd.notna(row.get("Main Category")) else ""
-        l1c = str(row.get("Level 1 Code") or "").strip() if pd.notna(row.get("Level 1 Code")) else ""
-        l1n = str(row.get("Level 1 Name") or "").strip() if pd.notna(row.get("Level 1 Name")) else ""
-        l2c = str(row.get("Level 2 Code") or "").strip() if pd.notna(row.get("Level 2 Code")) else ""
-        l2n = str(row.get("Level 2 Name") or "").strip() if pd.notna(row.get("Level 2 Name")) else ""
-        l3c = str(row.get("Level 3 Code") or "").strip() if pd.notna(row.get("Level 3 Code")) else ""
-        l3n = str(row.get("Level 3 Name") or "").strip() if pd.notna(row.get("Level 3 Name")) else ""
-        
-        if not l1c or not l1n:
-            continue
-            
-        parts = [f"{main_category} | {l1c} {l1n}"]
-        if l2c:
-            parts.append(f"{l2c} {l2n}")
-        if include_level3 and l3c:
-            parts.append(f"{l3c} {l3n}")
-        line = " | ".join(parts)
-        if line not in seen:
-            seen.add(line)
-            lines.append(line)
-    
-    return "\n".join(lines)
-
-
-# ============================ Image Analysis =====================================
-def extract_fsc_logo_and_taxonomy(image_path: str, product_type: str, taxonomy_block: str, 
-                                 client: openai.OpenAI = None, model_name: str = "gpt-4.1") -> Dict:
-    """
-    Describes image and material, matches taxonomy items, gets text
-    """
-    if client is None:
-        raise ValueError("OpenAI client must be provided")
-    
-    INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs(model_name)
-    
-    content = [{"type": "input_text", "text": prompt_image_analysis(product_type, taxonomy_block)}]
-
-    if image_path.lower().endswith(".pdf"):
-        image_paths = pdf_multi_page_to_images(image_path)
-    else:
-        return "Unsupported file format."
-
-    for image_path in image_paths:
-        img_b64 = file_to_b64(image_path)
-        content.append({"type": "input_image", "image_url": f"data:image/png;base64,{img_b64}", "detail": "high"})
-
-    logger.info(f"Image analysis content: {len(image_paths)} pages")
-
-    try:
-        response = client.responses.parse(
-            model=model_name, 
-            input=[{"role": "user", "content": content}],
-            text_format=FSCExtractionAndTaxonomy,
-        )
-
-        usage = response.usage
-        input_tokens = usage.input_tokens
-        output_tokens = usage.output_tokens
-        input_cost = (input_tokens / 1000.0) * INPUT_TOKEN_COST
-        output_cost = (output_tokens / 1000.0) * OUTPUT_TOKEN_COST
-        call_cost = input_cost + output_cost
-
-        logger.info(f"Image analysis: {input_tokens} in + {output_tokens} out = ${call_cost:.4f}")
-
-        if response.output_parsed:
-            result = response.output_parsed
-            result_dict = result.model_dump()
-
-            # flatten top-1 taxonomy into columns
-            tx = result_dict.get("taxonomy_matches", [])[:1]
-            if tx:
-                t0 = tx[0]
-                result_dict["main_category"] = t0.get("main_category")
-                result_dict["level1_code"] = t0.get("level1_code")
-                result_dict["level1_name"] = t0.get("level1_name")
-                result_dict["level2_code"] = t0.get("level2_code")
-                result_dict["level2_name"] = t0.get("level2_name")
-                result_dict["taxonomy_confidence"] = t0.get("confidence")
-            else:
-                # No taxonomy matches found
-                result_dict["main_category"] = None
-                result_dict["level1_code"] = None
-                result_dict["level1_name"] = None
-                result_dict["level2_code"] = None
-                result_dict["level2_name"] = None
-                result_dict["taxonomy_confidence"] = None
-            
-        return {
-            "image_data": result_dict,
-            "usage": {
-                "input_tokens": input_tokens,
-                "output_tokens": output_tokens,
-                "total_cost": call_cost
-            }
-        }
-
-    except Exception as e:
-        logger.error(f"Error processing Step 3: {str(e)}")
-        raise
-
-    finally:
-        # Clean up temporary PNG if created for single and multi page pdfs
-        for image_path in image_paths:
-            try:
-                os.remove(image_path)
-                logger.debug(f"Cleaned up temporary PNG: {image_path} for {len(image_paths)} pages")
-            except OSError:
-                logger.warning(f"Failed to clean up temporary PNG: {image_path} for {len(image_paths)} pages")
-
-
-
-# ============================ FSC Text Analysis ======================================
-
-# Analyze text from image and return all instances of the text "FSC" or "Forest Stewardship Council"
-def analyze_fsc_text(image_text: str) -> str:
-    """
-    Return all instances of "FSC" or "Forest Stewardship Council" with up to 3 words of context on each side.
-    Stops at newline characters.
-    """
-    if not image_text:
-        return ""
-
-    word = r"\b\w+(?:[-'']\w+)*\b"
-    pattern = rf"(?:{word}[^\w\n]+){{0,5}}(?:FSC|Forest Stewardship Council)(?:[^\w\n]+{word}){{0,5}}"
-    matches = re.findall(pattern, image_text, flags=re.IGNORECASE)
-    if not matches:
-        return "No FSC text strings identified"
-    
-    return "\n".join(f"{i}: {match}" for i, match in enumerate(matches, start=1))
-

modules/image_processing.py

@@ -4,7 +4,6 @@ import logging
 from pathlib import Path
 import pypdfium2 as pdfium
 from PIL import Image, ImageOps, ImageChops, ImageFilter
-from typing import List
 
 # Configure logging
 logger = logging.getLogger(__name__)
@@ -21,21 +20,15 @@ def _autocrop_white(im, tol=3):
     bbox = gray.point(lambda p: 255 if p > tol else 0).getbbox()
     return im.crop(bbox) if bbox else im
 
-
-# 
+# TO DO: fix this for multiple pages
 def pdf_first_page_to_image(
     pdf_path: str,
-    max_long_edge: int = 2000,        # 1600–2400 is a good sweet spot
+    max_long_edge: int = 2000,        # 1600–2400 is a good sweet spot for LLM vision
     fmt: str = "WEBP",                # "WEBP" | "JPEG" | "PNG"
     quality: int = 82,                # for WEBP/JPEG
     supersample: float = 1.7,         # render a bit larger, then downsample with LANCZOS
     crop_margins: bool = True
 ) -> str:
-    """
-    OpenAI API will not accept pdf as image files. 
-    Needs to be intelligently converted.
-    Using WEBP as default (offers the best tradeoff between compression and quality for this use case)
-    """
     pdf = pdfium.PdfDocument(pdf_path)
     page = pdf.get_page(0)
     w_pt, h_pt = page.get_size()      # PDF points
@@ -72,75 +65,4 @@ def pdf_first_page_to_image(
         img.save(out, "PNG", optimize=True, compress_level=9)
     else:
         raise ValueError("fmt must be one of: WEBP, JPEG, PNG")
-    return out
-
-def pdf_multi_page_to_images(
-    pdf_path: str,
-    max_pages: int = 3,  # Limit to first 3 pages to avoid excessive processing
-    max_long_edge: int = 2000,
-    fmt: str = "WEBP",
-    quality: int = 82,
-    supersample: float = 1.7,
-    crop_margins: bool = True
-) -> List[str]:
-    """
-    Convert PDF to multiple images (first N pages).
-    
-    Returns:
-        List[str]: List of temporary image file paths
-    """
-    pdf = pdfium.PdfDocument(pdf_path)
-    total_pages = len(pdf)
-    pages_to_process = min(max_pages, total_pages)
-    
-    image_paths = []
-    pdf_name = Path(pdf_path).stem
-    
-    for page_num in range(pages_to_process):
-        try:
-            page = pdf.get_page(page_num)
-            w_pt, h_pt = page.get_size()
-            long_pt = max(w_pt, h_pt)
-            
-            # Render bigger, then downsample
-            scale = supersample * (max_long_edge / long_pt)
-            pil = page.render(scale=scale).to_pil()
-            img = pil.convert("RGB")
-            
-            # Downsample to exact target
-            if max(img.size) > max_long_edge:
-                ratio = max_long_edge / max(img.size)
-                new_size = (int(img.width * ratio), int(img.height * ratio))
-                img = img.resize(new_size, Image.LANCZOS).filter(
-                    ImageFilter.UnsharpMask(radius=0.6, percent=120, threshold=3)
-                )
-            
-            if crop_margins:
-                img = _autocrop_white(img)
-            
-            # Create temporary file
-            with tempfile.NamedTemporaryFile(
-                suffix=f".{fmt.lower()}", 
-                delete=False, 
-                prefix=f"{pdf_name}_page{page_num+1}_"
-            ) as tmp_file:
-                out_path = tmp_file.name
-            
-            # Save with appropriate format
-            if fmt.upper() == "WEBP":
-                img.save(out_path, "WEBP", quality=quality, method=6)
-            elif fmt.upper() == "JPEG":
-                img.save(out_path, "JPEG", quality=quality, optimize=True, progressive=True, subsampling=0)
-            elif fmt.upper() == "PNG":
-                img = img.quantize(colors=256)
-                img.save(out_path, "PNG", optimize=True, compress_level=9)
-            else:
-                raise ValueError("fmt must be one of: WEBP, JPEG, PNG")
-            
-            image_paths.append(out_path)
-            
-        except Exception as e:
-            logger.error(f"Error processing page {page_num + 1}: {e}")
-            continue
-    
-    return image_paths
+    return out

modules/llm_pipeline.py

@@ -0,0 +1,433 @@
+import os
+import json
+import logging
+import re
+from typing import List, Optional, Dict
+import pandas as pd
+import openai
+from .models import FSCExtractionAndTaxonomy
+from .utils import get_token_costs
+from .image_processing import file_to_b64, pdf_first_page_to_image
+from .fsc_api import get_fsc_products_for_license, filter_taxonomy_by_fsc_products
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# ============================ Prompt constructors ===================================
+def build_taxonomy_block_from_dataframe(taxonomy_df: pd.DataFrame, include_level3: bool = False) -> str:
+    """
+    Produces compact lines from a filtered taxonomy dataframe such as:
+      W12 Indoor furniture | W12.1 Cabinet
+      P5 Packaging and wrappings of paper | P5.1 Cardboard packaging | P5.2 Corrugated paper packaging
+    """
+    lines, seen = [], set()
+    
+    for _, row in taxonomy_df.iterrows():
+        # Handle NaN values properly
+        l1c = str(row.get("Level 1 Code") or "").strip() if pd.notna(row.get("Level 1 Code")) else ""
+        l1n = str(row.get("Level 1 Name") or "").strip() if pd.notna(row.get("Level 1 Name")) else ""
+        l2c = str(row.get("Level 2 Code") or "").strip() if pd.notna(row.get("Level 2 Code")) else ""
+        l2n = str(row.get("Level 2 Name") or "").strip() if pd.notna(row.get("Level 2 Name")) else ""
+        l3c = str(row.get("Level 3 Code") or "").strip() if pd.notna(row.get("Level 3 Code")) else ""
+        l3n = str(row.get("Level 3 Name") or "").strip() if pd.notna(row.get("Level 3 Name")) else ""
+        
+        if not l1c or not l1n:
+            continue
+            
+        parts = [f"{l1c} {l1n}"]
+        if l2c:
+            parts.append(f"{l2c} {l2n}")
+        if include_level3 and l3c:
+            parts.append(f"{l3c} {l3n}")
+        line = " | ".join(parts)
+        if line not in seen:
+            seen.add(line)
+            lines.append(line)
+    
+    return "\n".join(lines)
+
+def build_prompt(taxonomy_block: str) -> str:
+    return f"""
+You are a compliance analyst assessing products for alignment with the Forest Stewardship Council (FSC) product taxonomy.
+
+TASKS:
+1. Describe what the product appears to be
+2. Infer the substrate/material
+3. Assess if the product matches any category in the provided taxonomy
+4. If there's a match, select 1–3 closest taxonomy alignments from the TAXONOMY below
+5. If there's NO match, set has_taxonomy_match to false and leave taxonomy_matches empty
+6. If you can't interpret what the product is from the image, set has_taxonomy_match to false and leave taxonomy_matches empty
+
+IMPORTANT: Only choose codes/names that appear in TAXONOMY. If the product doesn't clearly match any category in the taxonomy, indicate this by setting has_taxonomy_match to false.
+
+Heuristics / Few-shot hints:
+- Retail cereal boxes are folding cartons/cardboard → prefer P5.1 over P5.2 unless corrugation is explicit
+- Chests/commodes/cupboards/drawers → W12 Indoor furniture → W12.1 Cabinet
+- If the product appears to be something not covered by the taxonomy (e.g., electronics, clothing, food items), set has_taxonomy_match to false
+
+Often the image will features pictures(s) of other non-relevant products (e.g. bicycles). The actual FSC product can in such cases be a printed material or other substrate (e.g. packaging, electronic media). Carefully assess what the product is, taking this into account.
+When in doubt,arefully read all text in the image to help infer what the product is vs. what the actual FSC product is.
+A major flag is when you see a schematic which indicates a packaging design document. In this case, the product indicated on the packaging may differ from the actual FSC product.
+Also be careful as some images are taken from a screenshot of a website. In this case, read all text as it may give an indication of the FSC product type.
+
+Return JSON that matches the provided schema.
+
+## TAXONOMY (codes and names only; choose from these if there's a match)
+{taxonomy_block}
+
+Now analyze this image:
+""".strip()
+
+# ============================ Core extractor =====================================
+def extract_fsc_logo_and_taxonomy(image_path: str, taxonomy_block: str, original_filename: str = None, 
+                                 client: openai.OpenAI = None, model_name: str = "gpt-5", 
+                                 total_tokens: Dict[str, int] = None, total_cost: List[float] = None):
+    """
+    Single vision call that performs both:
+      - FSC OCR extraction
+      - Product taxonomy classification
+    """
+    if client is None:
+        raise ValueError("OpenAI client must be provided")
+    if total_tokens is None:
+        total_tokens = {"input": 0, "output": 0}
+    if total_cost is None:
+        total_cost = [0.0]
+    
+    INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs(model_name)
+
+    # Use original filename if provided, otherwise fall back to image_path basename
+    display_name = original_filename if original_filename else os.path.basename(image_path)
+    logger.info(f"Processing: {display_name}")
+    
+    img_b64 = file_to_b64(image_path)
+    content = [{"type": "text", "text": build_prompt(taxonomy_block)}]
+
+    content += [{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}", "detail": "high"}}]
+
+    try:
+        response = client.beta.chat.completions.parse(
+            model=model_name,
+            messages=[{"role": "user", "content": content}],
+            response_format=FSCExtractionAndTaxonomy,
+        )
+
+        usage = response.usage
+        input_tokens = usage.prompt_tokens
+        output_tokens = usage.completion_tokens
+        input_cost = (input_tokens / 1000.0) * INPUT_TOKEN_COST
+        output_cost = (output_tokens / 1000.0) * OUTPUT_TOKEN_COST
+        call_cost = input_cost + output_cost
+
+        total_tokens["input"] += input_tokens
+        total_tokens["output"] += output_tokens
+        total_cost[0] += call_cost
+
+        logger.info(f"✓ {display_name}: {input_tokens} in + {output_tokens} out = ${call_cost:.4f}")
+
+        if response.choices and response.choices[0].message and response.choices[0].message.parsed:
+            result = response.choices[0].message.parsed
+            result_dict = result.model_dump()
+            result_dict["source_file"] = display_name
+
+            # Convenience: flatten top-1 taxonomy into columns
+            tx = result_dict.get("taxonomy_matches", [])[:1]
+            if tx:
+                t0 = tx[0]
+                result_dict["level1_code"] = t0.get("level1_code")
+                result_dict["level1_name"] = t0.get("level1_name")
+                result_dict["level2_code"] = t0.get("level2_code")
+                result_dict["level2_name"] = t0.get("level2_name")
+                result_dict["taxonomy_confidence"] = t0.get("confidence")
+            else:
+                # No taxonomy matches found
+                result_dict["level1_code"] = None
+                result_dict["level1_name"] = None
+                result_dict["level2_code"] = None
+                result_dict["level2_name"] = None
+                result_dict["taxonomy_confidence"] = None
+            
+            return result_dict
+
+        # Fallback if no parseable output returned
+        logger.warning(f"⚠ No parseable output for {display_name}")
+        return {
+            "source_file": display_name,
+            "product_summary": None,
+            "inferred_material": None,
+            "taxonomy_matches": [],
+            "has_taxonomy_match": False,
+            "level1_code": None,
+            "level1_name": None,
+            "level2_code": None,
+            "level2_name": None,
+            "taxonomy_confidence": None,
+        }
+    except Exception as e:
+        logger.error(f"✗ Error processing {display_name}: {str(e)}")
+        raise
+
+# ============================ Main / Batch ======================================
+def load_fsc_license_mapping(fsc_logo_extracted_csv: str) -> Dict[str, str]:
+    """
+    Load FSC license codes from the extracted CSV file.
+    
+    Args:
+        fsc_logo_extracted_csv (str): Path to the FSC extraction CSV file
+    
+    Returns:
+        Dict[str, str]: Mapping of source_file to fsc_license_code
+    """
+    try:
+        df = pd.read_csv(fsc_logo_extracted_csv)
+        # Create mapping from source_file to fsc_license_code
+        mapping = {}
+        for _, row in df.iterrows():
+            source_file = row.get('source_file', '')
+            license_code = row.get('fsc_license_code', '')
+            if pd.notna(license_code) and license_code:
+                mapping[source_file] = license_code
+        
+        logger.info(f"Loaded {len(mapping)} FSC license mappings")
+        
+        # Debug: Show some examples of the mapping
+        sample_keys = list(mapping.keys())[:5]
+        logger.info(f"Sample mapping keys: {sample_keys}")
+        
+        return mapping
+    except Exception as e:
+        logger.error(f"Error loading FSC license mapping: {e}")
+        return {}
+
+def get_files_to_process(directory: str, fsc_license_mapping: Dict[str, str]) -> List[str]:
+    """
+    Get list of files to process by matching CSV entries with actual files in directory.
+    
+    Args:
+        directory (str): Directory containing image files
+        fsc_license_mapping (Dict[str, str]): Mapping of source_file to fsc_license_code
+        
+    Returns:
+        List[str]: List of filenames to process
+    """
+    # Get all files in directory
+    all_files = [f for f in os.listdir(directory) if f.lower().endswith(('.pdf', '.png', '.jpg', '.jpeg', '.webp')) and not f.endswith('.page1.webp')]
+    
+    files_to_process = []
+    unmatched_csv_entries = []
+    
+    # For each CSV entry, try to find a matching file
+    for csv_filename in fsc_license_mapping.keys():
+        matching_file = find_matching_license_key(csv_filename, {f: f for f in all_files})
+        if matching_file:
+            files_to_process.append(matching_file)
+            logger.debug(f"Matched CSV entry '{csv_filename}' to file '{matching_file}'")
+        else:
+            unmatched_csv_entries.append(csv_filename)
+            logger.warning(f"No matching file found for CSV entry: {csv_filename}")
+    
+    # Remove duplicates (in case multiple CSV entries match the same file)
+    files_to_process = list(set(files_to_process))
+    
+    logger.info(f"Found {len(files_to_process)} files to process out of {len(fsc_license_mapping)} CSV entries")
+    logger.info(f"Unmatched CSV entries: {len(unmatched_csv_entries)}")
+    
+    if unmatched_csv_entries:
+        logger.info(f"Sample unmatched entries: {unmatched_csv_entries[:5]}")
+    
+    return files_to_process
+
+def find_matching_license_key(filename: str, license_mapping: Dict[str, str]) -> Optional[str]:
+    """
+    Find a matching license key for a given filename by trying different variations.
+    
+    Args:
+        filename (str): The filename with extension (e.g., "KM548644-fsc.pdf")
+        license_mapping (Dict[str, str]): The license mapping dictionary
+        
+    Returns:
+        Optional[str]: The matching license key or None if not found
+    """
+    # Try exact match first
+    if filename in license_mapping:
+        return filename
+    
+    # Try without extension
+    filename_no_ext = os.path.splitext(filename)[0]
+    if filename_no_ext in license_mapping:
+        return filename_no_ext
+    
+    # Try with different common extensions
+    for ext in ['.pdf', '.png', '.jpg', '.jpeg', '.webp']:
+        test_filename = filename_no_ext + ext
+        if test_filename in license_mapping:
+            return test_filename
+    
+    # Try removing common suffixes that might have been added during processing
+    # Remove hash suffixes like "_2d4b076e5c1f4db6893970e23a14809c"
+    filename_clean = re.sub(r'_[a-f0-9]{32}$', '', filename_no_ext)
+    if filename_clean in license_mapping:
+        return filename_clean
+    
+    # Try removing hash suffixes with different patterns
+    filename_clean2 = re.sub(r'_[a-f0-9]{8,}$', '', filename_no_ext)
+    if filename_clean2 in license_mapping:
+        return filename_clean2
+    
+    return None
+
+def process_directory_with_fsc_filtering(directory: str, taxonomy_df: pd.DataFrame, 
+                                       fsc_license_mapping: Dict[str, str], 
+                                       client: openai.OpenAI, model_name: str = "gpt-5") -> pd.DataFrame:
+    logger.info(f"Starting processing of directory: {directory}")
+    
+    # Get files to process based on CSV entries
+    files_to_process = get_files_to_process(directory, fsc_license_mapping)
+    
+    if not files_to_process:
+        logger.error("No files to process found!")
+        return pd.DataFrame()
+    
+    # Debug: Show some examples of files to process
+    sample_files = files_to_process[:5]
+    logger.info(f"Sample files to process: {sample_files}")
+    
+    rows = []
+    total_tokens = {"input": 0, "output": 0}
+    total_cost = [0.0]
+    
+    for i, fname in enumerate(files_to_process, 1):
+        fpath = os.path.join(directory, fname)
+        lower = fname.lower()
+
+        logger.info(f"Processing file {i}/{len(files_to_process)}: {fname}")
+
+        # Get FSC license code for this file using flexible matching
+        matching_key = find_matching_license_key(fname, fsc_license_mapping)
+        fsc_license_code = fsc_license_mapping.get(matching_key) if matching_key else None
+        
+        if not fsc_license_code:
+            logger.warning(f"No FSC license code found for {fname} (tried matching key: {matching_key})")
+            # Debug: Show what keys are available for similar filenames
+            filename_no_ext = os.path.splitext(fname)[0]
+            similar_keys = [k for k in fsc_license_mapping.keys() if filename_no_ext in k or k in filename_no_ext]
+            if similar_keys:
+                logger.info(f"Similar keys found: {similar_keys[:3]}")  # Show first 3
+            continue
+        
+        logger.info(f"Found FSC license code: {fsc_license_code} for file: {fname}")
+        
+        # Get allowed products from FSC API
+        logger.info(f"Fetching FSC products for license: {fsc_license_code}")
+        fsc_products, license_status = get_fsc_products_for_license(fsc_license_code)
+        
+        # Check if license status is not 'Valid'
+        if license_status and license_status != 'Valid':
+            logger.warning(f"License status is '{license_status}' for {fsc_license_code}, skipping taxonomy processing")
+            # Add a row with license status and skip taxonomy processing
+            rows.append({
+                "source_file": fname,
+                "fsc_license_code": fsc_license_code,
+                "license_status": license_status,
+                "product_summary": None,
+                "inferred_material": None,
+                "has_taxonomy_match": False,
+                "level1_code": None,
+                "level1_name": None,
+                "level2_code": None,
+                "level2_name": None,
+                "taxonomy_confidence": None,
+                "all_taxonomy_matches": "[]",
+                "fsc_products_count": len(fsc_products),
+                "fsc_products": json.dumps(fsc_products, ensure_ascii=False),
+            })
+            continue
+        
+        if not fsc_products:
+            logger.warning(f"No FSC products found for license {fsc_license_code}, using full taxonomy")
+            filtered_taxonomy = taxonomy_df
+        else:
+            # Filter taxonomy based on FSC products
+            filtered_taxonomy = filter_taxonomy_by_fsc_products(taxonomy_df, fsc_products)
+            
+            if filtered_taxonomy.empty:
+                logger.warning(f"No matching taxonomy found for FSC products, using full taxonomy")
+                filtered_taxonomy = taxonomy_df
+        
+        # Build taxonomy block from filtered taxonomy
+        taxonomy_block = build_taxonomy_block_from_dataframe(filtered_taxonomy, include_level3=True)
+        
+        if not taxonomy_block.strip():
+            logger.warning(f"Empty taxonomy block for {fname}, using full taxonomy")
+            taxonomy_block = build_taxonomy_block_from_dataframe(taxonomy_df, include_level3=True)
+
+        # Convert PDF to temporary PNG (first page), otherwise use image as-is
+        cleanup_png = False
+        try:
+            if lower.endswith(".pdf"):
+                img_path = pdf_first_page_to_image(fpath)
+                cleanup_png = True
+                # Log the size of the converted PNG file
+                converted_file_size = os.path.getsize(img_path) / (1024 * 1024)  # Convert to MB
+                logger.info(f"Converted PNG size: {converted_file_size:.2f} MB")
+            elif lower.endswith((".png", ".jpg", ".jpeg", ".webp")):
+                img_path = fpath
+            else:
+                continue  # skip unsupported
+
+            # Pass the original filename to ensure it's used in the results
+            out = extract_fsc_logo_and_taxonomy(img_path, taxonomy_block, original_filename=fname,
+                                              client=client, model_name=model_name,
+                                              total_tokens=total_tokens, total_cost=total_cost)
+
+            # Create a single row for each file
+            rows.append({
+                "source_file": out.get("source_file"),
+                "fsc_license_code": fsc_license_code,
+                "license_status": license_status,
+                "product_summary": out.get("product_summary"),
+                "inferred_material": out.get("inferred_material"),
+                "has_taxonomy_match": out.get("has_taxonomy_match", False),
+                "level1_code": out.get("level1_code"),
+                "level1_name": out.get("level1_name"),
+                "level2_code": out.get("level2_code"),
+                "level2_name": out.get("level2_name"),
+                "taxonomy_confidence": out.get("taxonomy_confidence"),
+                "all_taxonomy_matches": json.dumps(out.get("taxonomy_matches", []), ensure_ascii=False),
+                "fsc_products_count": len(fsc_products),
+                "fsc_products": json.dumps(fsc_products, ensure_ascii=False),
+            })
+
+        except Exception as e:
+            logger.error(f"Failed to process {fname}: {e}")
+            # Add a row with error information
+            rows.append({
+                "source_file": fname,
+                "fsc_license_code": fsc_license_code,
+                "license_status": license_status,
+                "product_summary": f"ERROR: {str(e)}",
+                "inferred_material": None,
+                "has_taxonomy_match": False,
+                "level1_code": None,
+                "level1_name": None,
+                "level2_code": None,
+                "level2_name": None,
+                "taxonomy_confidence": None,
+                "all_taxonomy_matches": "[]",
+                "fsc_products_count": len(fsc_products),
+                "fsc_products": json.dumps(fsc_products, ensure_ascii=False),
+            })
+        finally:
+            # Clean up temp PNG
+            if cleanup_png and 'img_path' in locals():
+                try:
+                    os.remove(img_path)
+                    logger.debug(f"Cleaned up temporary PNG: {img_path}")
+                except OSError:
+                    logger.warning(f"Failed to clean up temporary PNG: {img_path}")
+
+    df = pd.DataFrame(rows)
+    logger.info(f"Completed processing. Generated {len(df)} rows")
+    logger.info(f"Total tokens in/out: {total_tokens['input']}/{total_tokens['output']}")
+    logger.info(f"Total cost: ${total_cost[0]:.4f}")
+    return df

modules/logo_extraction.py

@@ -1,434 +0,0 @@
-#!/usr/bin/env python3
-"""
-Searches for FSC logo inside PDF pages using block-wise template matching + feature matching (SIFT/ORB).
-
-Behavior notes:
-- The code will use fitz (PyMuPDF) or pdf2image depending on PDF_2_IMG flag.
-- If a logo is found it is written to: OUTPUT_FOLDER/{pdf_filename}_logo.png
-- If OUTPUT_FOLDER does not exist, behavior is unchanged (file write will fail as before).
-- Print messages remain the same and continue to use flush=True.
-
-Optimizations:
-- Parallel block processing using ProcessPoolExecutor
-- Template feature caching across PDFs
-- Configurable detector (ORB default for speed, SIFT optional)
-- Coarse-to-fine search strategy
-- Reduced feature counts for better performance
-- Early termination on first match
-- Retry if failure
-
-Handles white-on-dark vs black-on-white logo problem by:
-1. Caching both original AND inverted template features
-2. Trying both during matching
-3. Selecting the best match automatically
-"""
-
-import os
-from typing import Any, Optional, Tuple, List
-from functools import partial
-from concurrent.futures import ProcessPoolExecutor, as_completed
-import cv2
-import fitz  # PyMuPDF
-import numpy as np
-from pdf2image import convert_from_path
-from PIL import Image
-import logging
-
-# import from modules
-from modules.homography_functions import serialize_keypoints, deserialize_keypoints, extract_logo_from_pdf_page_dual_features
-
-logger = logging.getLogger(__name__)
-
-Image.MAX_IMAGE_PIXELS = 933120000
-
-# ==== Global Constants ====
-TEMPLATE_PATH = os.path.join("assets","fsc-logo-black-and-white.jpg")
-PDF_FOLDER = os.path.join("template_matching","pdfs")
-OUTPUT_FOLDER = os.path.join("template_matching","extracted")
-PADDING_SCALE = 4
-DPI = 300
-PDF_2_IMG = False
-
-# Optimized detector settings
-USE_SIFT = True  # Changed to False - ORB is much faster and often sufficient
-SIFT_FEATURES = 2500  # Reduced from 5000 for better performance
-ORB_FEATURES = 1500   # Increased from 1000 for better matching
-
-# Parallelization settings
-MAX_WORKERS = 4  # Adjust based on your CPU cores
-USE_COARSE_TO_FINE = False  # Enable coarse-to-fine search for large pages
-
-# Block processing settings
-OVERLAP = 0.2
-MIN_BLOCK_SIZE = 300
-MAX_BLOCK_SIZE = 1000
-# ==== End Global Constants ====
-
-# Global template cache
-_template_cache = {}
-
-
-
-def calculate_adaptive_block_size(page_shape: Tuple[int, int], 
-                                  min_size: int = MIN_BLOCK_SIZE, 
-                                  max_size: int = MAX_BLOCK_SIZE) -> Tuple[int, int]:
-    """Calculate block size as a fraction of image dimensions."""
-    H, W = page_shape[:2]
-    size = int(min(H, W) * 0.25)
-    size = max(min_size, min(size, max_size))
-    return (size, size)
-
-
-def get_template_features(template_path: str, use_sift: bool = USE_SIFT):
-    """
-    Load and cache BOTH original and inverted template features.
-    
-    This allows the system to match logos regardless of polarity
-    (black-on-white vs white-on-black).
-    
-    Returns
-    -------
-    Tuple containing:
-        - gray_template: grayscale template (original)
-        - kp_orig: keypoints from original
-        - des_orig: descriptors from original
-        - kp_inv: keypoints from inverted
-        - des_inv: descriptors from inverted
-        - detector: the feature detector used
-    """
-    cache_key = (template_path, use_sift)
-    
-    if cache_key not in _template_cache:
-        logger.info(f"Computing template features (SIFT={use_sift}) for BOTH polarities...")
-        
-        # Load and resize template
-        template = cv2.imread(template_path)
-        gray_template = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY)
-        
-        h, w = gray_template.shape
-        max_dim = 300
-        if max(h, w) > max_dim:
-            scale = max_dim / max(h, w)
-            new_w = int(w * scale)
-            new_h = int(h * scale)
-            gray_template = cv2.resize(gray_template, (new_w, new_h), 
-                                      interpolation=cv2.INTER_AREA)
-            logger.info(f"Resized template from {w}x{h} to {new_w}x{new_h}")
-        
-        # Create detector
-        if use_sift:
-            detector = cv2.SIFT_create(nfeatures=SIFT_FEATURES)
-        else:
-            detector = cv2.ORB_create(nfeatures=ORB_FEATURES)
-        
-        # Extract features from ORIGINAL template
-        kp_orig, des_orig = detector.detectAndCompute(gray_template, None)
-        logger.info(f"Original template keypoints: {len(kp_orig)}")
-        
-        # Extract features from INVERTED template
-        gray_template_inv = 255 - gray_template
-        kp_inv, des_inv = detector.detectAndCompute(gray_template_inv, None)
-        logger.info(f"Inverted template keypoints: {len(kp_inv)}")
-        
-        # Cache both versions
-        _template_cache[cache_key] = (
-            gray_template, 
-            kp_orig, des_orig,
-            kp_inv, des_inv,
-            detector
-        )
-    
-    return _template_cache[cache_key]
-
-
-def process_single_block(block_data: Tuple, 
-                         gray_template: np.ndarray,
-                         serialized_kp_orig: List[Tuple],
-                         des_orig: np.ndarray,
-                         serialized_kp_inv: List[Tuple],
-                         des_inv: np.ndarray,
-                         use_sift: bool) -> Optional[Tuple]:
-    """
-    Process a single block with dual-template matching.
-    
-    Returns
-    -------
-    Optional[Tuple]
-        (M_block, x_offset, y_offset, block_num) if logo found, None otherwise
-    """
-    block, x_offset, y_offset, block_num, total_blocks = block_data
-    
-    logger.debug(f"Processing block {block_num}/{total_blocks} at ({x_offset},{y_offset})")
-    
-    # Create detector inside worker (can't pickle cv2 objects)
-    if use_sift:
-        detector = cv2.SIFT_create(nfeatures=SIFT_FEATURES)
-    else:
-        detector = cv2.ORB_create(nfeatures=ORB_FEATURES)
-    
-    # Deserialize keypoints for both templates
-    kp_orig = deserialize_keypoints(serialized_kp_orig)
-    kp_inv = deserialize_keypoints(serialized_kp_inv)
-    
-    try:
-        # Use the improved dual-template matching function
-        M_block = extract_logo_from_pdf_page_dual_features(
-            block, gray_template, detector,
-            kp_orig, des_orig,
-            kp_inv, des_inv
-        )
-        
-        if M_block is not None:
-            logger.info(f"✓ Logo found in block {block_num} at ({x_offset}, {y_offset})")
-            return (M_block, x_offset, y_offset, block_num)
-    
-    except Exception as e:
-        logger.warning(f"Block {block_num} error: {e}")
-    
-    return None
-
-
-def extract_with_homography(M_block: np.ndarray,
-                            x_offset: int,
-                            y_offset: int,
-                            page_cv: np.ndarray,
-                            gray_template: np.ndarray) -> np.ndarray:
-    """Extract logo region using homography matrix."""
-    # Transform block coordinates to page coordinates
-    T = np.array([[1, 0, x_offset], [0, 1, y_offset], [0, 0, 1]], dtype=np.float32)
-    M_page = T @ M_block
-    
-    # Warp template corners to page coords
-    h, w = gray_template.shape
-    pts = np.float32([[0, 0], [0, h], [w, h], [w, 0]]).reshape(-1, 1, 2)
-    dst = cv2.perspectiveTransform(pts, M_page)
-    
-    # Compute bounding box with padding
-    x_min, y_min = np.int32(dst.min(axis=0).ravel())
-    x_max, y_max = np.int32(dst.max(axis=0).ravel())
-    
-    x_pad = int((x_max - x_min) * PADDING_SCALE)
-    y_pad = int((y_max - y_min) * PADDING_SCALE)
-    
-    x1_crop = max(0, x_min - x_pad)
-    y1_crop = max(0, y_min - y_pad)
-    x2_crop = min(page_cv.shape[1], x_max + x_pad)
-    y2_crop = min(page_cv.shape[0], y_max + y_pad)
-    
-    return page_cv[y1_crop:y2_crop, x1_crop:x2_crop]
-
-
-def search_logo_in_blocks_parallel(page_cv: np.ndarray,
-                                    gray_template: np.ndarray,
-                                    detector: Any,
-                                    kp_orig: Any,
-                                    des_orig: Any,
-                                    kp_inv: Any,
-                                    des_inv: Any,
-                                    max_workers: int = MAX_WORKERS) -> Optional[np.ndarray]:
-    """
-    Parallel block-wise logo search with dual-template features.
-    """
-    H, W = page_cv.shape[:2]
-    bh, bw = calculate_adaptive_block_size(page_cv.shape)
-    step_h = int(bh * (1 - OVERLAP))
-    step_w = int(bw * (1 - OVERLAP))
-    
-    logger.info(f"Page: {W}x{H}, Block: {bw}x{bh}, Template: {gray_template.shape[1]}x{gray_template.shape[0]}")
-    
-    # Prepare all blocks upfront
-    blocks_data = []
-    block_num = 0
-    for y in range(0, H, step_h):
-        for x in range(0, W, step_w):
-            y1, y2 = y, min(y + bh, H)
-            x1, x2 = x, min(x + bw, W)
-            block = page_cv[y1:y2, x1:x2].copy()
-            block_num += 1
-            blocks_data.append((block, x1, y1, block_num, len(blocks_data)))
-    
-    total_blocks = len(blocks_data)
-    logger.info(f"Processing {total_blocks} blocks in parallel (workers={max_workers})")
-    
-    # Serialize template features for BOTH polarities
-    serialized_kp_orig = serialize_keypoints(kp_orig)
-    serialized_kp_inv = serialize_keypoints(kp_inv)
-    use_sift = isinstance(detector, cv2.SIFT)
-    
-    # Process blocks in parallel
-    with ProcessPoolExecutor(max_workers=max_workers) as executor:
-        process_func = partial(
-            process_single_block,
-            gray_template=gray_template,
-            serialized_kp_orig=serialized_kp_orig,
-            des_orig=des_orig,
-            serialized_kp_inv=serialized_kp_inv,
-            des_inv=des_inv,
-            use_sift=use_sift
-        )
-        
-        # Submit all blocks
-        futures = {executor.submit(process_func, bd): bd for bd in blocks_data}
-        
-        # Process results as they complete (early termination)
-        for future in as_completed(futures):
-            result = future.result()
-            
-            if result is not None:
-                M_block, x_offset, y_offset, block_num = result
-                
-                # Cancel remaining futures
-                for f in futures:
-                    if not f.done():
-                        f.cancel()
-                
-                logger.info(f"Early termination - logo found, cancelled remaining blocks")
-                
-                # Extract logo with homography
-                return extract_with_homography(
-                    M_block, x_offset, y_offset, page_cv, gray_template
-                )
-    
-    logger.info("No logo found in any block")
-    return None
-
-
-def search_logo_with_retry(page_cv: np.ndarray,
-                          gray_template: np.ndarray,
-                          detector: Any,
-                          kp_orig: Any,
-                          des_orig: Any,
-                          kp_inv: Any,
-                          des_inv: Any,
-                          max_retries: int = 3) -> Optional[np.ndarray]:
-    """Search for logo with retry mechanism using dual templates."""
-    for attempt in range(max_retries):
-        logger.info(f"Logo search attempt {attempt + 1}/{max_retries}")
-        
-        result = search_logo_in_blocks_parallel(
-            page_cv, gray_template, detector, 
-            kp_orig, des_orig,
-            kp_inv, des_inv
-        )
-        
-        if result is not None:
-            logger.info(f"Logo found on attempt {attempt + 1}")
-            return result
-        
-        if attempt < max_retries - 1:
-            logger.info(f"Attempt {attempt + 1} failed, retrying...")
-    
-    logger.info("All retry attempts failed")
-    return None
-
-
-def process_pdf(pdf_path: str,
-                gray_template: np.ndarray,
-                detector: Any,
-                kp_orig: Any,
-                des_orig: Any,
-                kp_inv: Any,
-                des_inv: Any) -> Optional[np.ndarray]:
-    """
-    Process a PDF file with dual-template matching.
-    
-    Returns
-    -------
-    Optional[np.ndarray]
-        Cropped logo image if found, None otherwise
-    """
-    if PDF_2_IMG:
-        pages = convert_from_path(pdf_path, dpi=DPI)
-    else:
-        pages = fitz.open(pdf_path)
-    
-    filename = os.path.splitext(os.path.basename(pdf_path))[0]
-    
-    for page_num, page in enumerate(pages, 1):
-        try:
-            logger.info(f"Processing page {page_num} of {filename}")
-            
-            if PDF_2_IMG:
-                page_cv = cv2.cvtColor(np.array(page), cv2.COLOR_RGB2BGR)
-            else:
-                pix = page.get_pixmap(dpi=DPI)
-                img = np.frombuffer(pix.samples, dtype=np.uint8).reshape(
-                    pix.height, pix.width, pix.n
-                )
-                if pix.n == 4:
-                    page_cv = cv2.cvtColor(img, cv2.COLOR_RGBA2BGR)
-                else:
-                    page_cv = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
-            
-            # Search with dual-template retry
-            logo = search_logo_with_retry(
-                page_cv, gray_template, detector,
-                kp_orig, des_orig,
-                kp_inv, des_inv,
-                max_retries=3
-            )
-            
-            if logo is not None:
-                logger.info(f"Logo found on page {page_num}")
-                return logo
-            else:
-                logger.info(f"No logo on page {page_num}")
-                
-        except Exception as e:
-            logger.error(f"Error processing page {page_num} of {filename}: {e}")
-    
-    return None
-
-
-def get_crop_from_pdf(pdf: Optional[str]) -> Optional[str]:
-    """
-    Main entry point with polarity-invariant processing.
-    
-    Parameters
-    ----------
-    pdf : Optional[str]
-        Path to PDF file
-        
-    Returns
-    -------
-    Optional[str]
-        Path to extracted logo image, or None if not found
-    """
-    if pdf is None:
-        return (
-            "Please upload a pdf file.",
-            "⏳ Waiting for pdf...",
-            "⏳ Waiting for pdf...",
-            "⏳ Waiting for pdf...",
-            "⏳ Waiting for pdf...",
-            "",
-        )
-    
-    # Get cached template features (both polarities)
-    gray_template, kp_orig, des_orig, kp_inv, des_inv, detector = get_template_features(
-        TEMPLATE_PATH, USE_SIFT
-    )
-    
-    if pdf.lower().endswith(".pdf"):
-        logger.info(f"Processing PDF: {pdf}")
-        logger.info(f"Using {'SIFT' if USE_SIFT else 'ORB'} detector with {MAX_WORKERS} workers")
-        logger.info("Polarity-invariant matching enabled (white-on-dark logos supported)")
-        
-        cropped_logo = process_pdf(
-            pdf, gray_template, detector,
-            kp_orig, des_orig,
-            kp_inv, des_inv
-        )
-        
-        if cropped_logo is not None:
-            os.makedirs("assets/tmp", exist_ok=True)
-            output_path = "assets/tmp/cropped_logo.png"
-            cv2.imwrite(output_path, cropped_logo)
-            logger.info(f"✓ Logo saved to {output_path}")
-            return output_path
-        else:
-            logger.info("No logo found")
-            return None
-    else:
-        logger.warning("File is not a PDF")
-        return None

modules/models.py

@@ -1,18 +1,7 @@
 from pydantic import BaseModel, Field
-from typing import List, Optional
+from typing import List, Optional, Dict
 
-# OCR Extraction 
-class FSCLogoData(BaseModel):
-    label_title: Optional[str] = Field(description="Short text from the logo's Label Title field, e.g. 'MIX', 'RECYCLED', '100%'")
-    product_type: Optional[str] = Field(description="Descriptive text from the logo's 'Product Type' field")
-    fsc_license_code: Optional[str] = Field(description="The 7 character FSC license code from the logo. The code always starts with a C followed by 6 digits. Return null if you can't see a full 6 digit code.")
-
-class FSCExtractionResult(BaseModel):
-    fsc_data: List[FSCLogoData]
-
-# Image analysis
 class TaxonomyMatch(BaseModel):
-    main_category: Optional[str] = None
     level1_code: Optional[str] = None
     level1_name: Optional[str] = None
     level2_code: Optional[str] = None
@@ -21,17 +10,16 @@ class TaxonomyMatch(BaseModel):
     rationale: Optional[str] = None
 
 class FSCExtractionAndTaxonomy(BaseModel):
-    product_summary: str = Field(description="Describe the product")
-    inferred_material: str = Field(description="Describe the material/substrate of the product")
+    product_summary: Optional[str] = None
+    inferred_material: Optional[str] = None
     taxonomy_matches: List[TaxonomyMatch]
     has_taxonomy_match: bool = Field(description="Whether the product matches any category in the provided taxonomy")
-    image_text: Optional[str] = Field(description="Get all text from the image")
 
+# OCR Extraction Models
+class FSCLogoData(BaseModel):
+    label_title: Optional[str] = Field(description="The title of the label, e.g. 'MIX', RECYCLED' etc.")
+    product_type: Optional[str] = Field(description="The type of product, e.g. 'Paper', 'Wood', 'Packaging'")
+    fsc_license_code: Optional[str] = Field(description="The 7 character FSC license code. The code always starts with a C followed by 6 digits")
 
-# Product type check
-class ProductTypeMatch(BaseModel):
-    has_material_check: bool = Field(description="Whether the material approximately aligns with the product type")
-    material_rationale: Optional[str] = Field(description="The rationale for the material match")
-    has_taxonomy_check: bool = Field(description="Whether one or more of the taxonomy items approximately align with the product type")
-    taxonomy_rationale: Optional[str] = Field(description="The rationale for the taxonomy match")
-    taxonomy_matches: List[TaxonomyMatch] = Field(description="The taxonomy items that approximately align with the product type")
+class FSCExtractionResult(BaseModel):
+    fsc_data: List[FSCLogoData]

modules/ocr_extraction.py

@@ -1,77 +1,17 @@
 import os
 import base64
+import json
 import logging
+from typing import Dict, Optional, List
 import openai
-from rapidfuzz import fuzz, process
-
-# import from modules
-from modules.models import FSCExtractionResult
-from modules.prompts import prompt_ocr
-from modules.utils import get_token_costs
+from .models import FSCExtractionResult
+from .utils import get_token_costs
 
+# Configure logging
 logger = logging.getLogger(__name__)
 
-# Valid FSC label titles for fuzzy matching
-VALID_LABEL_TITLES = [
-    "100%",
-    "Mix",
-    "Recycled",
-]
-
-def correct_label_title(extracted_title: str, threshold: int = 85) -> str:
-    """
-    Correct OCR errors in label titles using fuzzy matching.
-    
-    Args:
-        extracted_title: The label title extracted by OCR
-        threshold: Minimum similarity score (0-100) to accept a match
-        
-    Returns:
-        Corrected label title or original if no good match found
-    """
-    if not extracted_title or extracted_title == "Not Found":
-        return extracted_title
-    
-    # Normalize the input
-    normalized_title = extracted_title.strip()
-    
-    # Check for exact match first (case-insensitive)
-    for valid_title in VALID_LABEL_TITLES:
-        if normalized_title.upper() == valid_title.upper():
-            return valid_title
-    
-    # Use fuzzy matching to find the best match (case-insensitive)
-    # Convert to uppercase for comparison
-    normalized_upper = normalized_title.upper()
-    valid_titles_upper = [title.upper() for title in VALID_LABEL_TITLES]
-    
-    result = process.extractOne(
-        normalized_upper,
-        valid_titles_upper,
-        scorer=fuzz.ratio,
-        score_cutoff=threshold
-    )
-    
-    if result:
-        matched_title_upper, score, matched_index = result
-        # Return the original casing from VALID_LABEL_TITLES
-        matched_title = VALID_LABEL_TITLES[matched_index]
-        
-        if normalized_title != matched_title:
-            logger.info(
-                f"OCR correction: '{normalized_title}' → '{matched_title}' "
-                f"(similarity: {score}%)"
-            )
-        return matched_title
-    
-    # No good match found, return original
-    logger.warning(
-        f"No fuzzy match found for label title: '{normalized_title}' "
-        f"(threshold: {threshold}%)"
-    )
-    return extracted_title
-
-def extract_fsc_logo_data_single(image_path: str, client: openai.OpenAI, model_name: str = "gpt-4.1") -> dict:
+def extract_fsc_logo_data_single(image_path: str, client: openai.OpenAI, model_name: str = "gpt-4o", 
+                                total_tokens: Dict[str, int] = None, total_cost: List[float] = None) -> dict:
     """
     Extract FSC logo data from a single image file.
     
@@ -79,73 +19,188 @@ def extract_fsc_logo_data_single(image_path: str, client: openai.OpenAI, model_n
         image_path: Path to the image file
         client: OpenAI client instance
         model_name: Model to use for extraction
+        total_tokens: Dictionary to track token usage
+        total_cost: List to track total cost
         
     Returns:
-        Dictionary with extracted FSC data and usage information
+        Dictionary with extracted FSC data
     """
-
-    INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs(model_name)
+    if total_tokens is None:
+        total_tokens = {"input": 0, "output": 0}
+    if total_cost is None:
+        total_cost = [0.0]
+    
+    try:
+        INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs(model_name)
+    except ValueError:
+        # Fallback to gpt-4o pricing if model not found
+        logger.warning(f"Model {model_name} not found in pricing table, using gpt-4o pricing")
+        INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs("gpt-4o")
     
     # Read and encode the target image file
     with open(image_path, "rb") as img_file:
         img_bytes = img_file.read()
         img_base64 = base64.b64encode(img_bytes).decode('utf-8')
     
-    # Try to read example image if it exists
-    example_image_path = os.path.join("assets","guidance.png")
-    if not os.path.exists(example_image_path):
-        logger.error("Can't find OCR guidance image")
-        return None
-        
-    # Read and encode example image
-    with open(example_image_path, "rb") as example_file:
-        example_bytes = example_file.read()
-        example_base64 = base64.b64encode(example_bytes).decode('utf-8')
-
-    content = [{"type": "input_text", "text": prompt_ocr()},
-    {"type": "input_image", "image_url": f"data:image/png;base64,{example_base64}", "detail": "high"},
-    {"type": "input_text", "text": "Now analyze this image and extract all FSC logo data:"},
-    {"type": "input_image", "image_url": f"data:image/png;base64,{img_base64}", "detail": "high"}]
-
-    response = client.responses.parse(
-        model=model_name, 
-        input=[{"role": "user", "content": content}],
-        text_format=FSCExtractionResult,
-    )
-
-    # Calculate usage and cost
+    # Try to read example image if it exists, otherwise use a simpler prompt
+    example_image_path = "guidance.png"
+    if os.path.exists(example_image_path):
+        # Read and encode an example image (one shotting works better)
+        with open(example_image_path, "rb") as example_file:
+            example_bytes = example_file.read()
+            example_base64 = base64.b64encode(example_bytes).decode('utf-8')
+
+        # Enhanced prompt with example
+        prompt = """
+You are an expert at detecting and extracting data from FSC (Forest Stewardship Council) certification logos in images.
+
+Here's an example of what an FSC logo looks like and the data you should extract:
+
+[EXAMPLE IMAGE WILL BE SHOWN BELOW]
+
+From this example FSC logo, you would extract:
+- Label Title: "RECYCLED"
+- Product Type: "Papier aus Recyclingmaterial"
+- FSC License Code: "C075521"
+
+Your Task:
+Extract three specific data fields:
+   - Label Title 
+   - Product Type 
+   - FSC License Code
+
+If no FSC logos are found, return all null values.
+
+Now analyze the following image and extract all FSC logo data:
+"""
+
+        # API call (includes example and target images)
+        response = client.beta.chat.completions.parse(
+            model=model_name,
+            messages=[
+                {
+                    "role": "user", 
+                    "content": [
+                        {"type": "text", "text": prompt},
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": f"data:image/png;base64,{example_base64}",
+                                "detail": "high"
+                            }
+                        },
+                        {"type": "text", "text": "Now analyze this image:"},
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": f"data:image/png;base64,{img_base64}",
+                                "detail": "high"
+                            }
+                        }
+                    ]
+                }
+            ],
+            response_format=FSCExtractionResult,
+        )
+    else:
+        # Simplified prompt without example image
+        prompt = """
+You are an expert at detecting and extracting data from FSC (Forest Stewardship Council) certification logos in images.
+
+Your Task:
+Extract three specific data fields from FSC logos in the image:
+   - Label Title (e.g., "MIX", "RECYCLED", "PURE")
+   - Product Type (e.g., "Paper", "Wood", "Packaging")
+   - FSC License Code (7 character code starting with C followed by 6 digits, e.g., "C075521")
+
+Look for FSC certification logos, labels, or text in the image. If no FSC logos are found, return all null values.
+
+Now analyze the following image and extract all FSC logo data:
+"""
+
+        # API call (target image only)
+        response = client.beta.chat.completions.parse(
+            model=model_name,
+            messages=[
+                {
+                    "role": "user", 
+                    "content": [
+                        {"type": "text", "text": prompt},
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": f"data:image/png;base64,{img_base64}",
+                                "detail": "high"
+                            }
+                        }
+                    ]
+                }
+            ],
+            response_format=FSCExtractionResult,
+        )
+
+    # Track token usage and cost
     usage = response.usage
-    input_tokens = usage.input_tokens
-    output_tokens = usage.output_tokens
+    input_tokens = usage.prompt_tokens
+    output_tokens = usage.completion_tokens
+    
+    # Calculate costs
     input_cost = (input_tokens / 1000) * INPUT_TOKEN_COST
     output_cost = (output_tokens / 1000) * OUTPUT_TOKEN_COST
     call_cost = input_cost + output_cost
     
+    # Update totals
+    total_tokens["input"] += input_tokens
+    total_tokens["output"] += output_tokens
+    total_cost[0] += call_cost
+    
     logger.info(f"OCR extraction: {input_tokens} input + {output_tokens} output = ${call_cost:.4f}")
 
-    # Extract and return the result
-    result = response.output_parsed
-    result_dict = result.model_dump()
-    
-    for item in result_dict["fsc_data"]:
-        for key, value in item.items():
-            if value is None:
-                item[key] = "Not Found"
-
-        # Apply fuzzy matching correction to label_title
-        if "label_title" in item:
-            original_title = item["label_title"]
-            corrected_title = correct_label_title(original_title)
-            item["label_title"] = corrected_title
-    
-    return {
-        "fsc_data": result_dict["fsc_data"],
-        "usage": {
-            "input_tokens": input_tokens,
-            "output_tokens": output_tokens,
-            "total_cost": call_cost
+    # Extract the parsed result
+    if response.choices[0].message.parsed:
+        result = response.choices[0].message.parsed
+        
+        # Add source_file deterministically to each item
+        result_dict = result.model_dump()
+        for item in result_dict["fsc_data"]:
+            item["source_file"] = os.path.basename(image_path)
+        
+        return result_dict
+    else:
+        # Fallback case - create structure with source_file
+        fallback_result = {
+            "fsc_data": [{
+                "source_file": os.path.basename(image_path),
+                "label_title": None,
+                "product_type": None,
+                "fsc_license_code": None,
+            }]
         }
-    }
+        return fallback_result
+
+# Keep the original function for backward compatibility
+def extract_fsc_logo_data(directory: str, image_file: str, client: openai.OpenAI, 
+                         model_name: str = "gpt-4o", total_tokens: Dict[str, int] = None, 
+                         total_cost: List[float] = None) -> str:
+    """
+    Original function for batch processing - kept for backward compatibility.
+    
+    Args:
+        directory: Directory containing the image file
+        image_file: Name of the image file
+        client: OpenAI client instance
+        model_name: Model to use for extraction
+        total_tokens: Dictionary to track token usage
+        total_cost: List to track total cost
+        
+    Returns:
+        JSON string with extracted FSC data
+    """
+    image_path = os.path.join(directory, image_file)
+    result = extract_fsc_logo_data_single(image_path, client, model_name, total_tokens, total_cost)
+    return json.dumps(result)
+
+
 
 
 

modules/pipeline.py

@@ -1,305 +0,0 @@
-import os
-import logging
-from typing import List, Dict, Tuple
-import openai
-import configparser
-import pandas as pd
-
-# import from modules
-from modules.logo_extraction import get_crop_from_pdf
-from modules.ocr_extraction import extract_fsc_logo_data_single
-from modules.database_lookup import get_fsc_products_for_license, filter_taxonomy_by_fsc_products
-from modules.image_analysis import extract_fsc_logo_and_taxonomy, build_taxonomy_block_from_dataframe, analyze_fsc_text
-from modules.scope_check import check_product_type_match, step5_label_title_check
-from modules.utils import extract_cost
-
-
-# ============================ Config ============================================
-
-# Configure logging
-logger = logging.getLogger(__name__)
-
-# Load configuration
-config = configparser.ConfigParser()
-config.read('config.cfg')
-
-# Expect your API key in environment: OPENAI_API_KEY
-client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
-
-# Model configuration
-OCR_MODEL = config.get('ocr', 'MODEL')
-SCOPE_MODEL = config.get('scope', 'MODEL')
-
-# Load taxonomy
-try:
-    taxonomy_df = pd.read_csv(os.path.join("assets","product_classification.csv"))
-    logger.info(f"Loaded taxonomy with {len(taxonomy_df)} categories")
-except Exception as e:
-    logger.error(f"Failed to load taxonomy: {e}")
-    taxonomy_df = pd.DataFrame()
-
-# Import dictionary of label_title translations and package as dictionary for mix and recycled (100% is the same in all languages so not needed))
-try:
-    translations_df = pd.read_csv(os.path.join("assets","label_title_mapping.csv"))
-    logger.info(f"Loaded translations dictionary")
-    mix = set(translations_df['category_mix'].to_list())
-    recycled = set(translations_df['category_recycled'].to_list())
-    translations_dict = {"mix" : mix, "recycled" : recycled}
-except:
-    logger.error(f"Failed to load translations dictionary: {e}")
-
-# ============================ Pipeline ============================================
-
-def step1_extract_fsc_code(image_file) -> Tuple[str, str, str, str]:
-    """Step 1: Extract FSC license code using OCR."""
-    
-    try:
-        image_path = image_file
-        # Step 1: Extract FSC license code using OCR extraction
-        logger.info("Step 1: Extracting FSC license code using OCR...")
-            
-        # Use the updated OCR extraction function
-        ocr_data = extract_fsc_logo_data_single(
-            image_path, 
-            client, 
-            model_name=OCR_MODEL
-        )
-
-        fsc_entry = ocr_data["fsc_data"][0]
-        fsc_code = fsc_entry.get("fsc_license_code")
-        label_title = fsc_entry.get("label_title")
-        product_type = fsc_entry.get("product_type")
-
-        logger.info(f"OCR extracted FSC license code: {fsc_code}")
-
-        # Get usage information from the response
-        usage = ocr_data.get("usage", {})
-        total_cost = usage.get("total_cost", 0.0)
-        input_tokens = usage.get("input_tokens", 0)
-        output_tokens = usage.get("output_tokens", 0)
-        
-        # Format cost information
-        cost_info = f"""
-Total Cost: ${total_cost:.4f}
-Model Used: {OCR_MODEL}
-Token Usage: {input_tokens} input + {output_tokens} output tokens"""
-
-        if fsc_code == "Not Found":
-            return fsc_code, label_title, product_type, cost_info
-
-        # Ensure FSC- prefix
-        if not fsc_code.startswith("FSC-"):
-            fsc_code = f"FSC-{fsc_code}"
-
-        return fsc_code, label_title, product_type, cost_info
-            
-    except Exception as e:
-        logger.error(f"OCR extraction failed: {e}")
-        return f"Error in OCR extraction: {str(e)}"
-        
-
-def step2_fsc_lookup(fsc_code) -> Tuple[List[Dict], str, str, str, str, str]:
-    """Step 2: Lookup FSC license in API and get scope."""
-    if not fsc_code or fsc_code == "Not Found":
-        return [], "No FSC code found - skipping API lookup"
-    
-    try:
-        logger.info(f"Step 2: Looking up FSC license: {fsc_code}")
-        
-        fsc_products, license_status, certificate_status = get_fsc_products_for_license(fsc_code)
-        
-        # Handle case where license is not found in FSC database
-        if license_status == "Not Found":
-            logger.warning(f"FSC license {fsc_code} not found in FSC database")
-            return [], "FSC Code Not Found in Database", ""
-        
-        return fsc_products, license_status, certificate_status
-        
-    except Exception as e:
-        logger.error(f"Error in FSC lookup: {e}")
-        return [], "Error in FSC lookup"
-
-def step3_analyze_with_filtered_taxonomy(image_file, fsc_code, product_type, fsc_products, license_status) -> Tuple[str, str, str, str, str, str]:
-    """Step 3: Analyze product with FSC-filtered taxonomy."""
-    
-    try:
-        # Convert PDF to image if needed
-        cleanup_png = False
-        # Step 3: Filter taxonomy and analyze
-        logger.info("Step 3: Analyzing product with filtered taxonomy...")
-        
-        # Determine which taxonomy to use
-        if fsc_code != "Not Found" and license_status == 'Valid' and fsc_products:
-            # filter full taxonomy based on licensed FSC products
-            filtered_taxonomy = filter_taxonomy_by_fsc_products(taxonomy_df, fsc_products)
-            # Construct the filtered taxonomy to be LLM friendly (for prompt construction)
-            if not filtered_taxonomy.empty:
-                taxonomy_block = build_taxonomy_block_from_dataframe(filtered_taxonomy, include_level3=True)
-                taxonomy_source = "FSC-filtered"
-        # If no scope matches found, use full taxonomy
-        else:
-            # Use full taxonomy
-            taxonomy_block = build_taxonomy_block_from_dataframe(taxonomy_df, include_level3=True)
-            taxonomy_source = "full"
-        
-        logger.info(f"Using {taxonomy_source} taxonomy for analysis")
-        
-        result = extract_fsc_logo_and_taxonomy(
-            image_file,
-            product_type,
-            taxonomy_block, 
-            client=client, 
-            model_name=SCOPE_MODEL
-        )
-
-        image_data = result.get("image_data", {})
-        product_summary = image_data.get("product_summary", "No product summary available")
-        material = image_data.get("inferred_material", "No material information available")
-        
-        # Format taxonomy results
-        taxonomy_matches = image_data.get("taxonomy_matches", [])
-        has_match = image_data.get("has_taxonomy_match", False)
-        
-        taxonomy_results = ""
-        if has_match and taxonomy_matches:
-            for i, match in enumerate(taxonomy_matches[:3], 1):  # Show top 3 matches
-                taxonomy_results += f"Match {i}:\n"
-                taxonomy_results += f"Main Category: {match['main_category']}\n"
-                if match.get("level1_code") and match.get("level1_name"):
-                    taxonomy_results += f"Level 1: {match['level1_code']} - {match['level1_name']}\n"
-                if match.get("level2_code") and match.get("level2_name"):
-                    taxonomy_results += f"Level 2: {match['level2_code']} - {match['level2_name']}\n"
-                if match.get("confidence"):
-                    taxonomy_results += f"Confidence: {match['confidence']:.2f}\n"
-                if match.get("rationale"):
-                    taxonomy_results += f"Rationale: {match['rationale']}\n"
-                taxonomy_results += "\n"
-        else:
-            taxonomy_results += "Taxonomy Classification: No matching categories found"
-
-        image_text = image_data.get("image_text", "No image text available")
-
-                # Get usage information from the response
-        usage = result.get("usage", {})
-        total_cost = usage.get("total_cost", 0.0)
-        input_tokens = usage.get("input_tokens", 0)
-        output_tokens = usage.get("output_tokens", 0)
-
-        cost_info = f"""
-Total Cost: ${total_cost:.4f}
-Model Used: {SCOPE_MODEL}
-Token Usage: {input_tokens} input + {output_tokens} output tokens"""
-        
-        return product_summary, material, taxonomy_results, image_text, cost_info
-                    
-    except Exception as e:
-        logger.error(f"Error in step 3: {e}")
-
-def step4_product_type_check(product_type: str, material: str, taxonomy_results: str) -> Tuple[str, str]:
-    """Step 4: Check if the product type aligns with image analysis"""
-
-    # Call the scope_check module function
-    result = check_product_type_match(
-        product_type=product_type,
-        material=material,
-        taxonomy_results=taxonomy_results,
-        client=client,
-        model_name=SCOPE_MODEL
-    )
-
-    product_type_match = result.get("product_type_match", {})
-
-    usage = result.get("usage", {})
-    total_cost = usage.get("total_cost", 0.0)
-    input_tokens = usage.get("input_tokens", 0)
-    output_tokens = usage.get("output_tokens", 0)
-    
-    # Format cost information
-    cost_info = f"""
-Total Cost: ${total_cost:.4f}
-Model Used: {SCOPE_MODEL}
-Token Usage: {input_tokens} input + {output_tokens} output tokens"""
-    
-    return product_type_match, cost_info
-
-# ============================ Bring it all together ============================================
-
-def process_image_progressive(image_file):
-    """Process image with progressive updates showing each step."""
-
-     # Step 0: Search for FSC logo
-    if image_file.lower().endswith(".pdf"):
-        yield "⏳ Step 0: Searching for FSC Logo. Grab a coffee - this can take a while...", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""
-        image_file_cropped = get_crop_from_pdf(image_file)
-    else:
-        return "Please upload an image file in pdf format", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""
-    
-    # If we get a failed logo extraction, return an error
-    if not image_file_cropped:
-        yield "No logo found in the image", "", "", "", "", "", "", "", "", "Manual Review", "", "", "", "", "", ""
-        return
-    else:
-        yield "⏳ Step 1: Extracting Logo Data...", "⏳ Step 1: Extracting Logo Data...", "⏳ Step 1: Extracting Logo Data...", "", "", "", "", "", "", "", "", "", "", "", "", ""
-        
-        # ========== Step 1: Extract FSC code ==========
-        fsc_code, label_title, product_type, cost_info_1 = step1_extract_fsc_code(image_file_cropped)
-        yield fsc_code, label_title, product_type, "⏳ Step 2: Looking up FSC license...", "", "", "", "", "", "", "", "", "", "", "", ""
-
-        if fsc_code == "Not Found":
-            yield fsc_code, label_title, product_type, "No FSC Code. Cannot proceed with analysis...", "No FSC Code. Cannot proceed with analysis...", "", "", "", "", "Manual Review", "", "", "", "", "", ""
-            return
-
-        # ========== Step 2: FSC lookup ==========
-        fsc_products, license_status, certificate_status = step2_fsc_lookup(fsc_code)
-
-        #Check if API call was good / license and certificate status is valid before proceeding to step 3
-        if license_status == "FSC Code Not Found in Database":
-            yield fsc_code, label_title, product_type, license_status, "", "", "", "", "", "Manual Review", "", "", "", "", "", ""
-            return
-        elif license_status == "" or certificate_status != "Valid":
-            license_status = "FSC Certificate Status: '"+certificate_status+"'"
-            yield fsc_code, label_title, product_type, license_status, "Analysis skipped due to invalid license and/or certificate status", "", "", "", "", "Manual Review", "", "", "", "", "", ""
-            return
-        elif license_status != "Valid":
-            yield fsc_code, label_title, product_type, license_status, "Analysis skipped due to invalid license status", "", "", "", "", "Manual Review", "", "", "", "", "", ""
-            return
-
-        yield fsc_code, label_title, product_type, license_status, "⏳ Step 3: Analyzing image...", "⏳ Step 3: Analyzing image...", "⏳ Step 3: Analyzing image...", "⏳ Step 3: Analyzing image...", "⏳ Step 3: Analyzing image...", "", "", "", "", "", "", ""
-
-
-        # ========== Step 3: Analyze image with filtered taxonomy ==========
-        product_summary, material, taxonomy_results, image_text, cost_info_3 = step3_analyze_with_filtered_taxonomy(image_file, fsc_code, product_type, fsc_products, license_status)
-        fsc_text_strings = analyze_fsc_text(image_text)
-        yield fsc_code, label_title, product_type, license_status, product_summary, material, taxonomy_results, image_text, fsc_text_strings, "⏳ Step 4: Checking Scope...", "⏳ Step 4: Checking Scope...", "⏳ Step 4: Checking Scope...", "⏳ Step 4: Checking Scope...", "⏳ Step 4: Checking Scope...", "⏳ Step 4: Checking Scope...",""
-        
-        # ========== Step 4: Check if the product type aligns with image analysis ==========
-        product_type_match, cost_info_4 = step4_product_type_check(product_type, material, taxonomy_results)
-    
-        # Add up the token cost for each step
-        cost_1 = extract_cost(cost_info_1)
-        cost_3 = extract_cost(cost_info_3)
-        cost_4 = extract_cost(cost_info_4)
-        total_cost_all_steps = cost_1 + cost_3 + cost_4
-
-        # Combine cost information from all three steps
-        combined_cost_info = f"""=== TOTAL COST (ALL STEPS) ===
-    ${total_cost_all_steps:.4f}
-
-    === STEP 1: OCR EXTRACTION ==={cost_info_1}
-
-    === STEP 3: IMAGE ANALYSIS ==={cost_info_3}
-
-    === STEP 4: SCOPE CHECK ==={cost_info_4}"""
-
-        # ========== Step 5: Check if the label title matches the taxonomy matches ==========
-        label_title_check = step5_label_title_check(label_title, fsc_products, product_type_match.taxonomy_matches, translations_dict)
-
-        # Get overall compliance status
-        overall_scope_check = "Ok" if (product_type_match.has_material_check and product_type_match.has_taxonomy_check and label_title_check) else "Manual Review"
-
-        product_type_check = "Ok" if product_type_match.has_material_check else "Not Ok"
-        taxonomy_check = "Ok" if product_type_match.has_taxonomy_check else "Not Ok"
-        label_title_check = "Ok" if label_title_check else "Not Ok"
-
-        yield fsc_code, label_title, product_type, license_status, product_summary, material, taxonomy_results, image_text, fsc_text_strings, overall_scope_check, product_type_check, taxonomy_check, label_title_check, product_type_match.material_rationale, product_type_match.taxonomy_rationale, combined_cost_info
-        return

modules/prompts.py

@@ -1,78 +0,0 @@
-# Repo for all three pipeline prompts
-
-def prompt_ocr():
-    return """
-You are an expert at detecting and extracting data from FSC (Forest Stewardship Council) certification logos in images.
-
-Your Task:
-Analyze images of FSC logos and extract three specific data fields:
-   - Label Title (short text label field)
-   - Product Type (text description field)
-   - FSC License Code (short text code field)
-
-If no FSC logos are found, return all null values.
-
-Please see example below of what an FSC logo looks like and the data you should extract. From this example FSC logo, you would extract:
-- Label Title: "RECYCLED"
-- Product Type: "Papier aus Recyclingmaterial"
-- FSC License Code: "C075521"
-
-Here is the example:
-""".strip()
-
-
-def prompt_image_analysis(product_type: str, taxonomy_block: str) -> str:
-    return f"""
-You are a compliance analyst assessing products for alignment with the Forest Stewardship Council (FSC) product taxonomy. You will be provided with an image of a product from an FSC licensee for approval.
-The product identification task is quite difficult - as it is not always clear from images whether the product pictured in the image is the actual licensed product, /
-or whether the image depicts a representation of another item *inside* the licensed product. 
-For example, a FSC licensee that makes paper-based packaging might submit a schematic of packaging design for woodent children's toy. As both wood and paper products (inter alia) are covered by FSC categories, /
-it is difficult to tell whether the product is the packaging or the wooden toy. In such cases, it can be helpful to review the Product Type presented on the logo. However, the Product Type can be wrong in some cases, /
-so it is important to think critically about all aspects of the image, alongside the Product Type. If the Product Type mentions 'Paper' products and the filtered TAXONOMY (see below) does NOT mention 'Wood' products, /
-then we can infer that the primary product is the actual paper packaging of the toy.
-
-Often the image will features pictures(s) of other non-relevant products (e.g. bicycles). The actual FSC product can in such cases be a printed material or other substrate (e.g. packaging, electronic media). Carefully assess what the product is, taking this into account.
-When in doubt, carefully read all text in the image to help infer what the product is vs. what the actual FSC product is.
-A major flag is when you see a schematic which indicates a packaging design document. In this case, the product indicated on the packaging may differ from the actual FSC product.
-Also be careful as some images are taken from a screenshot of a website. In this case, read all text as it may give an indication of the FSC product type.
-
-Product Type: "{product_type}"
-
-## TAXONOMY (Filtered for this FSC Licensee. Respond with main category, codes and names; choose from these if there's a match)
-{taxonomy_block}
-
-TASKS:
-1. Describe what the product pictured in the image appears to be 
-2. Infer the substrate/material of the primary product pictured in the image (where we infer the primary product as relates to the FSC TAXONOMY and Product Type)
-3. Assess if the product matches any category in the provided TAXONOMY (where we infer the primary product as relates to the FSC TAXONOMY and Product Type)
-4. If there's a match, select 1–3 closest TAXONOMY alignments from the TAXONOMY below
-5. If there's NO match, set has_taxonomy_match to false and leave taxonomy_matches empty
-6. If you can't interpret what the product is from the image, set has_taxonomy_match to false and leave taxonomy_matches empty
-7. Extract all text from the image and return it in the image_text field.
-
-IMPORTANT: Only choose codes/names that appear in TAXONOMY. If the product doesn't clearly match any category in the taxonomy, indicate this by setting has_taxonomy_match to false.
-
-Heuristics / Few-shot hints:
-- Retail cereal boxes are folding cartons/cardboard → prefer P5.1 over P5.2 unless corrugation is explicit
-- Chests/commodes/cupboards/drawers → W12 Indoor furniture → W12.1 Cabinet
-- If the product appears to be something not covered by the taxonomy (e.g., electronics, clothing, food items), set has_taxonomy_match to false (but make sure to consider what the actual primary product is)
-
-Return your response as per the provided schema.
-
-Now analyze the image:
-""".strip()
-
-
-def prompt_product_type_check(product_type: str, material: str, taxonomy_results: str) -> str:
-    return f"""
-You are a compliance analyst assessing product alignment in the context communications.
-
-TASKS:
-1. Does the substrate/material "{material}" approximately match the material mentioned in the product type: "{product_type}"? Note - we only care here about the material, not any claims about sourcing/sustainability etc. We should also be very careful that the match is quite clear. For instance, if the substrate/material is "Cardboard" with a product type of "Paper products source from sustainable forests", then it would be a match as cardboard is a clear paper product. However, if the substrate/material is "Cardboard" with a product type of "Wood source from sustainable forests", it would NOT be a match. Even though cardboard comes from wood, we are looking for closer matches.
-2. Include rationale for your answer.
-3. Does one or more of the products mentioned in the taxonomy results ("{taxonomy_results}") approximately align with the material mentioned in "{product_type}"? E.g. if the product type is "Paper", then the taxonomy items could be something related to "Paper products". Here the 'Main Category' is the main category of the product type.
-4. Return the taxonomy items that approximately align with the product type in a list.
-5. Include rationale for your answer. 
-
-Return your response as per the provided schema.
-""".strip()

modules/scope_check.py

@@ -1,153 +0,0 @@
-import logging
-import re
-import os
-import pandas as pd
-from typing import List, Dict, Union, Any
-import openai
-
-# import from modules
-from modules.models import ProductTypeMatch
-from modules.prompts import prompt_product_type_check
-from modules.utils import get_token_costs
-
-# Configure logging
-logger = logging.getLogger(__name__)
-
-# ============================ Product Type Check ======================================
-
-def check_product_type_match(product_type: str, material: str, taxonomy_results: str,
-                             client: openai.OpenAI = None, model_name: str = "gpt-4.1") -> ProductTypeMatch:
-    """
-    Check if the product type aligns with the inferred material and taxonomy results.
-    
-    Args:
-        product_type: The product type extracted from the FSC label
-        material: The inferred material from image analysis
-        taxonomy_results: The taxonomy classification results
-        client: OpenAI client instance
-        model_name: Model to use for the check
-        
-    Returns:
-        ProductTypeMatch: Structured result with match assessment and rationale
-    """
-    if client is None:
-        raise ValueError("OpenAI client must be provided")
-    
-    INPUT_TOKEN_COST, OUTPUT_TOKEN_COST = get_token_costs(model_name)
-    
-    logger.info(f"Checking product type match: {product_type}")
-    
-    prompt = prompt_product_type_check(product_type, material, taxonomy_results)
-    
-    try:
-        response = client.responses.parse(
-            model=model_name, 
-            input=[{"role": "user", "content": prompt}],
-            text_format=ProductTypeMatch,
-        )
-        
-        usage = response.usage
-        input_tokens = usage.input_tokens
-        output_tokens = usage.output_tokens
-        input_cost = (input_tokens / 1000.0) * INPUT_TOKEN_COST
-        output_cost = (output_tokens / 1000.0) * OUTPUT_TOKEN_COST
-        call_cost = input_cost + output_cost
-        
-        logger.info(f"✓ Product type check: {input_tokens} in + {output_tokens} out = ${call_cost:.4f}")
-        
-        product_type_match = response.output_parsed
-
-        return {
-            "product_type_match": product_type_match,
-            "usage": {
-                "input_tokens": input_tokens,
-                "output_tokens": output_tokens,
-                "total_cost": call_cost
-            }
-        }
-    
-    except Exception as e:
-        logger.error(f"✗ Error in product type check: {str(e)}")
-        raise
-        
-
-# ============================ Label Title Check ======================================
-
-def step5_label_title_check(
-    label_title: str,
-    fsc_products: Union[Dict[str, Any], List[Dict[str, Any]]],
-    taxonomy_matches: Any,
-    translations_dict: dict
-) -> bool:
-    """
-    Step 5: 
-    1. Check if extracted Label Title matches any of the Step 3 identified taxonomy items (which were already filtered per FSC code scope in Step 3 prior to feeding to LLM)
-    2. Compare Label Titles of matched taxonomy items with the extracted Label Title
-    3. Return True if the extracted Label Title matches the Label Title of any matched taxonomy item
-    """
-
-    if not label_title or not taxonomy_matches or not fsc_products:
-        return False
-
-    # --- Extract taxonomy codes robustly ---
-    codes: set[str] = set()
-
-    def _extract_from_str(s: str):
-        codes.update(re.findall(r"level1_code='([^']+)'", s))
-        codes.update(re.findall(r"level2_code='([^']+)'", s))
-
-    if isinstance(taxonomy_matches, str):
-        _extract_from_str(taxonomy_matches)
-    elif isinstance(taxonomy_matches, dict):
-        for k in ("level1_code", "level2_code"):
-            v = taxonomy_matches.get(k)
-            if v: codes.add(v)
-    elif isinstance(taxonomy_matches, (list, tuple, set)):
-        for m in taxonomy_matches:
-            if isinstance(m, dict):
-                for k in ("level1_code", "level2_code"):
-                    v = m.get(k)
-                    if v: codes.add(v)
-            else:
-                # dataclass / simple object
-                for k in ("level1_code", "level2_code"):
-                    v = getattr(m, k, None)
-                    if v: codes.add(v)
-    else:
-        _extract_from_str(str(taxonomy_matches))
-
-    if not codes:
-        logger.warning("Could extract from filtered taxonomy")
-        return False
-
-
-    logo_lt = label_title.strip().upper()
-
-    # --- Match products ---
-    for product in fsc_products:
-        level1_text = (product.get("ProductLevel1") or "").strip()
-        level2_text = (product.get("ProductLevel2") or "").strip()
-
-        level1_code = level1_text.split(" ", 1)[0] if level1_text else ""
-        level2_code = level2_text.split(" ", 1)[0] if level2_text else ""
-
-        if level1_code in codes or level2_code in codes:
-            # Get allowed label titles for license scope
-            scope_lt = (product.get("MainOutputCategory") or "")
-            # Parse string response (format is like 'FSC Mix; FSC 100%; FSC Recycled')
-            scope_lt_ls = [item.strip().replace('FSC ', '', 1) for item in scope_lt.split(";")]
-            logger.info(f"Scoped LTs: {scope_lt_ls}, Target LT: {logo_lt}")
-            # Lookup scoped label_titles in translation dictionary
-            for lt in scope_lt_ls:
-                # 100% is same in all languages so no need for dictionary lookup
-                if lt == '100%':
-                    if logo_lt == lt:
-                        return True
-                else:
-                    scope_lt_translated = translations_dict[lt.lower()]
-                    # Then check that logo LT is in scope (using all possible translations)
-                    if logo_lt in scope_lt_translated:
-                        return True
-
-    return False
-

modules/utils.py

@@ -1,5 +1,3 @@
-import re
-
 # Model pricing lookup table
 MODEL_PRICING = {
     "gpt-5": {"input": 1.25, "cached_input": 0.125, "output": 10.00},
@@ -34,9 +32,4 @@ def get_token_costs(model_name: str) -> tuple[float, float]:
     input_cost = pricing["input"] / 1000.0
     output_cost = pricing["output"] / 1000.0
     
-    return input_cost, output_cost
-
-# Extract costs from each step for total calculation
-def extract_cost(cost_str):
-    match = re.search(r'Total Cost: \$([0-9.]+)', cost_str)
-    return float(match.group(1)) if match else 0.0
+    return input_cost, output_cost

requirements.txt

@@ -6,7 +6,4 @@ requests
 pypdfium2
 pdf2image
 Pillow
-gradio
-opencv-python-headless
-pymupdf
-rapidfuzz
+gradio