diff --git "a/app.py" "b/app.py"
--- "a/app.py"
+++ "b/app.py"
@@ -30,6 +30,15 @@ import uuid
# Check if running on Hugging Face Spaces
IS_HF_SPACE = os.getenv("SPACE_ID") is not None
+# Get Hugging Face token from environment or Streamlit secrets
+HF_TOKEN = None
+try:
+ # Try Streamlit secrets first (for HF Spaces)
+ HF_TOKEN = st.secrets.get("HF_TOKEN", None)
+except:
+ # Fall back to environment variable
+ HF_TOKEN = os.getenv("HF_TOKEN", None)
+
# Streamlit and core libraries
import streamlit as st
import sqlite3
@@ -235,8 +244,13 @@ class AIExtractor:
self.use_transformers = self.setup_transformers() if not self.use_mistral else False
def setup_mistral(self):
- """Try to setup Mistral 7B model"""
+ """Try to setup Mistral 7B model with proper authentication"""
try:
+ # Check if we have HF token
+ if not HF_TOKEN:
+ st.warning("โ ๏ธ Hugging Face token not found. Add HF_TOKEN to secrets for Mistral access.")
+ return False
+
# Check if we're in a high-resource environment
import psutil
memory_gb = psutil.virtual_memory().total / (1024**3)
@@ -246,15 +260,20 @@ class AIExtractor:
return False
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+ from huggingface_hub import login
+
+ # Login with HF token
+ login(token=HF_TOKEN)
with st.spinner("๐ Loading Mistral 7B model (this may take a few minutes)..."):
- # Use quantized version for better performance
+ # Use the instruction-tuned model
model_name = "mistralai/Mistral-7B-Instruct-v0.1"
# Load with reduced precision for memory efficiency
self.mistral_tokenizer = AutoTokenizer.from_pretrained(
model_name,
- cache_dir=HF_CONFIG["cache_dir"]
+ cache_dir=HF_CONFIG["cache_dir"],
+ token=HF_TOKEN
)
self.mistral_model = AutoModelForCausalLM.from_pretrained(
@@ -262,7 +281,8 @@ class AIExtractor:
torch_dtype=torch.float16 if TORCH_AVAILABLE else None,
device_map="auto" if TORCH_AVAILABLE else None,
load_in_8bit=True, # Use 8-bit quantization
- cache_dir=HF_CONFIG["cache_dir"]
+ cache_dir=HF_CONFIG["cache_dir"],
+ token=HF_TOKEN
)
# Create pipeline
@@ -282,6 +302,7 @@ class AIExtractor:
return False
except Exception as e:
st.warning(f"โ ๏ธ Mistral 7B not available: {e}")
+ st.info("๐ก To use Mistral 7B: Add your Hugging Face token to secrets as 'HF_TOKEN'")
return False
def setup_transformers(self):
@@ -444,98 +465,3997 @@ Respond with JSON only: [/INST]"""
return self.extract_with_regex(text)
def extract_with_regex(self, text: str) -> InvoiceData:
- """Fallback regex extraction"""
+ """Enhanced regex extraction with better amount detection"""
invoice_data = InvoiceData()
invoice_data.processing_method = "regex"
- # Enhanced regex patterns
+ # Enhanced regex patterns with more comprehensive matching
patterns = {
'invoice_number': [
- r'invoice[#\s]*:?\s*([A-Z0-9\-_]+)',
- r'inv[#\s]*:?\s*([A-Z0-9\-_]+)',
- r'bill[#\s]*:?\s*([A-Z0-9\-_]+)',
- r'#([A-Z0-9\-_]{3,})'
+ r'invoice\s*(?:no|number|#)?\s*:?\s*([A-Z0-9\-_/]+)',
+ r'bill\s*(?:no|number|#)?\s*:?\s*([A-Z0-9\-_/]+)',
+ r'inv\s*(?:no|number|#)?\s*:?\s*([A-Z0-9\-_/]+)',
+ r'ref\s*(?:no|number|#)?\s*:?\s*([A-Z0-9\-_/]+)',
+ r'#\s*([A-Z0-9\-_/]{3,})',
+ r'(?:^|\s)([A-Z]{2,}\d{3,}|\d{3,}[A-Z]{2,})', # Common patterns like ABC123 or 123ABC
],
'amount': [
- r'total[:\s]*[\$โน]?([0-9,]+\.?\d*)',
- r'amount[:\s]*[\$โน]?([0-9,]+\.?\d*)',
- r'[\$โน]([0-9,]+\.?\d*)',
- r'([0-9,]+\.?\d*)\s*(?:dollars?|โน|USD|INR)'
- ],
- 'date': [
- r'date[:\s]*(\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4})',
- r'(\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4})',
- r'(\d{4}[/\-]\d{1,2}[/\-]\d{1,2})'
- ],
- 'quantity': [
- r'qty[:\s]*(\d+)',
- r'quantity[:\s]*(\d+)',
- r'(\d+)\s*(?:pcs?|units?|items?)'
- ]
- }
+ # Currency symbols with amounts
+ r'total\s*(?:amount)?\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+ r'amount\s*(?:due|paid|total)?\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+ r'grand\s*total\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+ r'net\s*(?:amount|total)\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+ r'sub\s*total\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+
+ # Currency symbols at the beginning
+ r'[\$โนยฃโฌ]\s*([0-9,]+\.?\d*)',
+
+ # Amounts at end of lines (common in invoices)
+ r'([0-9,]+\.?\d*)\s*[\$โนยฃโฌ]?\s*
+
+ def parse_date(self, date_str: str) -> str:
+ """Parse date to YYYY-MM-DD format"""
+ if not date_str:
+ return ""
- text_lower = text.lower()
+ formats = ['%Y-%m-%d', '%m/%d/%Y', '%d/%m/%Y', '%m-%d-%Y', '%d-%m-%Y', '%Y/%m/%d']
- # Extract using patterns
- for pattern_list in patterns['invoice_number']:
- match = re.search(pattern_list, text_lower, re.IGNORECASE)
- if match:
- invoice_data.invoice_number = match.group(1).upper()
- break
+ for fmt in formats:
+ try:
+ parsed_date = datetime.strptime(date_str, fmt)
+ return parsed_date.strftime('%Y-%m-%d')
+ except ValueError:
+ continue
- for pattern in patterns['amount']:
- match = re.search(pattern, text_lower, re.IGNORECASE)
- if match:
- try:
- amount_str = match.group(1).replace(',', '')
- invoice_data.amount = float(amount_str)
- break
- except ValueError:
- continue
+ return date_str
+
+# ===============================================================================
+# VECTOR STORE CLASS
+# ===============================================================================
+
+class VectorStore:
+ """Simplified vector store for Hugging Face Spaces"""
+
+ def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
+ self.embedding_model_name = embedding_model
+ self.vector_store_path = os.path.join(HF_CONFIG["data_dir"], "vectors.pkl")
+ self.metadata_path = os.path.join(HF_CONFIG["data_dir"], "metadata.pkl")
+ self.embedding_model = None
+ self.vectors = []
+ self.document_metadata = []
+ self.embedding_dimension = None
- for pattern in patterns['date']:
- match = re.search(pattern, text, re.IGNORECASE)
- if match:
- invoice_data.date = self.parse_date(match.group(1))
- break
+ self.setup_embedding_model()
+ self.load_vector_store()
+
+ def setup_embedding_model(self):
+ """Initialize the sentence transformer model"""
+ if not SENTENCE_TRANSFORMERS_AVAILABLE:
+ st.warning("โ ๏ธ Sentence Transformers not available. Vector search disabled.")
+ return
- for pattern in patterns['quantity']:
- match = re.search(pattern, text_lower, re.IGNORECASE)
- if match:
- try:
- invoice_data.quantity = int(match.group(1))
- break
- except ValueError:
- continue
+ try:
+ with st.spinner(f"Loading embedding model: {self.embedding_model_name}..."):
+ self.embedding_model = SentenceTransformer(
+ self.embedding_model_name,
+ cache_folder=HF_CONFIG["cache_dir"]
+ )
+
+ # Get embedding dimension
+ test_embedding = self.embedding_model.encode(["test"])
+ self.embedding_dimension = test_embedding.shape[0]
+
+ st.success(f"โ
Embedding model loaded: {self.embedding_model_name}")
+
+ except Exception as e:
+ st.error(f"โ Failed to load embedding model: {e}")
+ self.embedding_model = None
+
+ def load_vector_store(self):
+ """Load existing vector store"""
+ try:
+ if os.path.exists(self.vector_store_path) and os.path.exists(self.metadata_path):
+ with open(self.vector_store_path, 'rb') as f:
+ self.vectors = pickle.load(f)
+
+ with open(self.metadata_path, 'rb') as f:
+ self.document_metadata = pickle.load(f)
+
+ st.success(f"โ
Vector store loaded: {len(self.document_metadata)} documents")
+ else:
+ self.vectors = []
+ self.document_metadata = []
+ st.info("๐ New vector store initialized")
+
+ except Exception as e:
+ st.error(f"โ Error loading vector store: {e}")
+ self.vectors = []
+ self.document_metadata = []
+
+ def save_vector_store(self):
+ """Save vector store to disk"""
+ try:
+ with open(self.vector_store_path, 'wb') as f:
+ pickle.dump(self.vectors, f)
+
+ with open(self.metadata_path, 'wb') as f:
+ pickle.dump(self.document_metadata, f)
+
+ return True
+ except Exception as e:
+ st.error(f"Error saving vector store: {e}")
+ return False
+
+ def create_document_text(self, invoice_data: dict, raw_text: str = "") -> str:
+ """Create searchable text from invoice data"""
+ text_parts = []
- # Extract company names (basic)
- company_patterns = [
- r'(?:from|to|vendor|supplier)[:]\s*([A-Z][A-Za-z\s&,\.]{2,30})',
- r'([A-Z][A-Za-z\s&,\.]{3,30})\s*(?:Ltd|Inc|Corp|LLC|Co\.|Company)',
- ]
+ for field, value in invoice_data.items():
+ if value and field != 'id':
+ text_parts.append(f"{field}: {value}")
- for pattern in company_patterns:
- matches = re.findall(pattern, text)
- if matches:
- if not invoice_data.supplier_name:
- invoice_data.supplier_name = matches[0].strip()
- elif len(matches) > 1 and not invoice_data.buyer_name:
- invoice_data.buyer_name = matches[1].strip()
-
- # Extract product description (basic)
- desc_patterns = [
- r'description[:]\s*([A-Za-z0-9\s,.-]{10,100})',
- r'item[:]\s*([A-Za-z0-9\s,.-]{10,100})',
- r'service[:]\s*([A-Za-z0-9\s,.-]{10,100})'
- ]
+ if raw_text:
+ text_parts.append(f"content: {raw_text[:300]}")
- for pattern in desc_patterns:
- match = re.search(pattern, text, re.IGNORECASE)
- if match:
- invoice_data.product_description = match.group(1).strip()
- break
+ return " | ".join(text_parts)
+
+ def add_document(self, invoice_data: dict, raw_text: str = "") -> bool:
+ """Add a document to the vector store"""
+ if not self.embedding_model:
+ return False
+
+ try:
+ document_text = self.create_document_text(invoice_data, raw_text)
+
+ # Generate embedding
+ embedding = self.embedding_model.encode(document_text, normalize_embeddings=True)
+
+ # Create metadata
+ metadata = {
+ 'invoice_id': invoice_data.get('id', ''),
+ 'invoice_number': invoice_data.get('invoice_number', ''),
+ 'supplier_name': invoice_data.get('supplier_name', ''),
+ 'buyer_name': invoice_data.get('buyer_name', ''),
+ 'amount': invoice_data.get('amount', 0),
+ 'date': invoice_data.get('date', ''),
+ 'file_name': invoice_data.get('file_info', {}).get('file_name', ''),
+ 'document_text': document_text[:200],
+ 'timestamp': datetime.now().isoformat()
+ }
+
+ # Add to store
+ self.vectors.append(embedding)
+ self.document_metadata.append(metadata)
+
+ return True
+
+ except Exception as e:
+ st.error(f"Error adding document to vector store: {e}")
+ return False
+
+ def semantic_search(self, query: str, top_k: int = 5) -> List[VectorSearchResult]:
+ """Perform semantic search using cosine similarity"""
+ if not self.embedding_model or not self.vectors:
+ return []
+
+ try:
+ # Generate query embedding
+ query_embedding = self.embedding_model.encode(query, normalize_embeddings=True)
+
+ # Calculate similarities
+ similarities = []
+ for i, doc_embedding in enumerate(self.vectors):
+ similarity = np.dot(query_embedding, doc_embedding)
+ similarities.append((similarity, i))
+
+ # Sort by similarity
+ similarities.sort(reverse=True)
+
+ # Return top results
+ results = []
+ for similarity, idx in similarities[:top_k]:
+ if similarity > 0.1: # Relevance threshold
+ metadata = self.document_metadata[idx]
+ result = VectorSearchResult(
+ invoice_id=metadata.get('invoice_id', ''),
+ invoice_number=metadata.get('invoice_number', ''),
+ supplier_name=metadata.get('supplier_name', ''),
+ similarity_score=float(similarity),
+ content_preview=metadata.get('document_text', ''),
+ metadata=metadata
+ )
+ results.append(result)
+
+ return results
+
+ except Exception as e:
+ st.error(f"Error in semantic search: {e}")
+ return []
+
+# ===============================================================================
+# MAIN PROCESSOR CLASS
+# ===============================================================================
+
+class InvoiceProcessor:
+ """Main invoice processor for Hugging Face Spaces"""
+
+ def __init__(self):
+ self.setup_storage()
+ self.document_processor = DocumentProcessor()
+ self.ai_extractor = AIExtractor()
+ self.vector_store = VectorStore() if SENTENCE_TRANSFORMERS_AVAILABLE else None
+
+ # Initialize stats
+ self.processing_stats = {
+ 'total_processed': 0,
+ 'successful': 0,
+ 'failed': 0,
+ 'start_time': datetime.now()
+ }
+
+ def setup_storage(self):
+ """Setup storage paths"""
+ self.data_dir = HF_CONFIG["data_dir"]
+ self.json_path = os.path.join(self.data_dir, "invoices.json")
+
+ # Initialize JSON storage
+ if not os.path.exists(self.json_path):
+ initial_data = {
+ "metadata": {
+ "created_at": datetime.now().isoformat(),
+ "version": "hf_v1.0",
+ "total_invoices": 0
+ },
+ "invoices": [],
+ "summary": {
+ "total_amount": 0.0,
+ "unique_suppliers": [],
+ "processing_stats": {"successful": 0, "failed": 0}
+ }
+ }
+ self.save_json_data(initial_data)
+
+ def load_json_data(self) -> dict:
+ """Load invoice data from JSON"""
+ try:
+ with open(self.json_path, 'r', encoding='utf-8') as f:
+ return json.load(f)
+ except (FileNotFoundError, json.JSONDecodeError):
+ self.setup_storage()
+ return self.load_json_data()
+
+ def save_json_data(self, data: dict):
+ """Save invoice data to JSON"""
+ try:
+ with open(self.json_path, 'w', encoding='utf-8') as f:
+ json.dump(data, f, indent=2, ensure_ascii=False)
+ except Exception as e:
+ st.error(f"Error saving data: {e}")
+
+ def process_uploaded_file(self, uploaded_file) -> InvoiceData:
+ """Process a single uploaded file with enhanced debugging"""
+ self.processing_stats['total_processed'] += 1
+
+ try:
+ # Debug file info
+ file_size = len(uploaded_file.getvalue())
+ file_extension = uploaded_file.name.split('.')[-1].lower() if '.' in uploaded_file.name else 'unknown'
+
+ st.info(f"๐ Processing: {uploaded_file.name} ({file_size/1024:.1f} KB, .{file_extension})")
+
+ # Check file size
+ if file_size > HF_CONFIG["max_file_size_mb"] * 1024 * 1024:
+ error_msg = f"File too large: {file_size / 1024 / 1024:.2f}MB > {HF_CONFIG['max_file_size_mb']}MB"
+ st.error(error_msg)
+ self.processing_stats['failed'] += 1
+ return InvoiceData()
+
+ # Check file type
+ if file_extension not in ['pdf', 'txt']:
+ error_msg = f"Unsupported file type: .{file_extension} (supported: PDF, TXT)"
+ st.warning(error_msg)
+ self.processing_stats['failed'] += 1
+ return InvoiceData()
+
+ # Save temporarily
+ with tempfile.NamedTemporaryFile(delete=False, suffix=f".{file_extension}") as tmp_file:
+ file_content = uploaded_file.getvalue()
+ tmp_file.write(file_content)
+ tmp_file_path = tmp_file.name
+
+ st.info(f"๐พ Saved temporarily to: {tmp_file_path}")
+
+ try:
+ # Extract text
+ st.info("๐ Extracting text from document...")
+ text = self.document_processor.extract_text_from_document(tmp_file_path)
+
+ if not text or not text.strip():
+ st.warning(f"โ No text extracted from {uploaded_file.name}")
+ self.processing_stats['failed'] += 1
+ return InvoiceData()
+
+ text_length = len(text)
+ st.info(f"๐ Extracted {text_length} characters of text")
+
+ # Show text preview and extraction debug info
+ if text_length > 0:
+ with st.expander("๐ Text Preview & Extraction Debug", expanded=True):
+ st.text_area("Extracted Text (First 1000 chars):", value=text[:1000], height=150, disabled=True)
+
+ # Debug amount detection
+ st.markdown("**๐ Amount Detection Debug:**")
+ amount_patterns = [
+ r'total\s*(?:amount)?\s*:?\s*[\$โนยฃโฌ]?\s*([0-9,]+\.?\d*)',
+ r'[\$โนยฃโฌ]\s*([0-9,]+\.?\d*)',
+ r'([0-9,]+\.?\d*)\s*[\$โนยฃโฌ]?\s*
+
+ # Extract invoice data
+ st.info("๐ค Extracting invoice data using AI/Regex...")
+ invoice_data = self.ai_extractor.extract_with_ai(text)
+ invoice_data.file_path = uploaded_file.name
+
+ # Show extraction results
+ st.info(f"๐ Extraction completed with {invoice_data.extraction_confidence:.1%} confidence")
+
+ # Save to storage
+ st.info("๐พ Saving extracted data...")
+ self.save_invoice_data(invoice_data, text, file_size)
+
+ self.processing_stats['successful'] += 1
+ st.success(f"โ
Successfully processed {uploaded_file.name}")
+
+ return invoice_data
+
+ finally:
+ # Cleanup
+ try:
+ os.unlink(tmp_file_path)
+ st.info("๐งน Cleaned up temporary file")
+ except:
+ pass
+
+ except Exception as e:
+ error_msg = f"Error processing {uploaded_file.name}: {str(e)}"
+ st.error(error_msg)
+ self.processing_stats['failed'] += 1
+
+ # Show detailed error for debugging
+ with st.expander("๐ Error Details", expanded=False):
+ st.code(str(e))
+ import traceback
+ st.code(traceback.format_exc())
+
+ return InvoiceData()
+
+ def save_invoice_data(self, invoice_data: InvoiceData, raw_text: str, file_size: int):
+ """Save invoice data to JSON and vector store"""
+ try:
+ # Load existing data
+ data = self.load_json_data()
+
+ # Create invoice record
+ invoice_record = {
+ "id": len(data["invoices"]) + 1,
+ "invoice_number": invoice_data.invoice_number,
+ "supplier_name": invoice_data.supplier_name,
+ "buyer_name": invoice_data.buyer_name,
+ "date": invoice_data.date,
+ "amount": invoice_data.amount,
+ "quantity": invoice_data.quantity,
+ "product_description": invoice_data.product_description,
+ "file_info": {
+ "file_name": invoice_data.file_path,
+ "file_size": file_size
+ },
+ "extraction_info": {
+ "confidence": invoice_data.extraction_confidence,
+ "method": invoice_data.processing_method,
+ "raw_text_preview": raw_text[:300]
+ },
+ "timestamps": {
+ "created_at": datetime.now().isoformat()
+ }
+ }
+
+ # Add to invoices
+ data["invoices"].append(invoice_record)
+
+ # Update summary
+ self.update_summary(data)
+
+ # Save JSON
+ self.save_json_data(data)
+
+ # Add to vector store
+ if self.vector_store:
+ self.vector_store.add_document(invoice_record, raw_text)
+ self.vector_store.save_vector_store()
+
+ except Exception as e:
+ st.error(f"Error saving invoice data: {e}")
+
+ def update_summary(self, data: dict):
+ """Update summary statistics"""
+ invoices = data["invoices"]
+
+ total_amount = sum(inv.get("amount", 0) for inv in invoices)
+ unique_suppliers = list(set(inv.get("supplier_name", "") for inv in invoices if inv.get("supplier_name")))
+
+ data["summary"] = {
+ "total_amount": total_amount,
+ "unique_suppliers": unique_suppliers,
+ "processing_stats": {
+ "successful": self.processing_stats['successful'],
+ "failed": self.processing_stats['failed'],
+ "total_processed": self.processing_stats['total_processed']
+ }
+ }
+
+ data["metadata"]["last_updated"] = datetime.now().isoformat()
+ data["metadata"]["total_invoices"] = len(invoices)
+
+# ===============================================================================
+# CHATBOT CLASS
+# ===============================================================================
+
+class ChatBot:
+ """Chatbot for invoice queries"""
+
+ def __init__(self, processor: InvoiceProcessor):
+ self.processor = processor
+
+ def query_database(self, query: str) -> str:
+ """Process user query and return response"""
+ try:
+ data = self.processor.load_json_data()
+ invoices = data.get("invoices", [])
+
+ if not invoices:
+ return "No invoice data found. Please upload some invoices first."
+
+ query_lower = query.lower()
+
+ # Handle different query types
+ if any(phrase in query_lower for phrase in ["summary", "overview", "total"]):
+ return self.generate_summary(data)
+
+ elif "count" in query_lower or "how many" in query_lower:
+ return self.handle_count_query(data)
+
+ elif any(phrase in query_lower for phrase in ["amount", "value", "money", "cost"]):
+ return self.handle_amount_query(data)
+
+ elif any(phrase in query_lower for phrase in ["supplier", "vendor", "company"]):
+ return self.handle_supplier_query(data, query)
+
+ elif self.processor.vector_store:
+ return self.handle_semantic_search(query)
+
+ else:
+ return self.handle_general_query(data, query)
+
+ except Exception as e:
+ return f"Error processing query: {e}"
+
+ def generate_summary(self, data: dict) -> str:
+ """Generate comprehensive summary"""
+ invoices = data.get("invoices", [])
+ summary = data.get("summary", {})
+
+ if not invoices:
+ return "No invoices found in the system."
+
+ total_amount = summary.get("total_amount", 0)
+ avg_amount = total_amount / len(invoices) if invoices else 0
+ unique_suppliers = len(summary.get("unique_suppliers", []))
+
+ response = f"""
+**๐ Invoice System Summary**
+
+โข **Total Invoices**: {len(invoices):,}
+โข **Total Value**: โน{total_amount:,.2f}
+โข **Average Invoice**: โน{avg_amount:,.2f}
+โข **Unique Suppliers**: {unique_suppliers}
+
+**๐ Processing Stats**
+โข **Successful**: {summary.get('processing_stats', {}).get('successful', 0)}
+โข **Failed**: {summary.get('processing_stats', {}).get('failed', 0)}
+
+**๐ Recent Invoices**
+"""
+
+ # Show recent invoices
+ recent = sorted(invoices, key=lambda x: x.get('timestamps', {}).get('created_at', ''), reverse=True)[:5]
+ for i, inv in enumerate(recent, 1):
+ response += f"\n{i}. **{inv.get('invoice_number', 'N/A')}** - {inv.get('supplier_name', 'Unknown')} (โน{inv.get('amount', 0):,.2f})"
+
+ return response
+
+ def handle_count_query(self, data: dict) -> str:
+ """Handle count-related queries"""
+ invoices = data.get("invoices", [])
+ total = len(invoices)
+ unique_numbers = len(set(inv.get('invoice_number', '') for inv in invoices if inv.get('invoice_number')))
+
+ return f"""
+**๐ Invoice Count Summary**
+
+โข **Total Records**: {total}
+โข **Unique Invoice Numbers**: {unique_numbers}
+โข **Duplicates**: {total - unique_numbers if total > unique_numbers else 0}
+
+**๐
Processing Timeline**
+โข **First Invoice**: {invoices[0].get('timestamps', {}).get('created_at', 'N/A')[:10] if invoices else 'N/A'}
+โข **Latest Invoice**: {invoices[-1].get('timestamps', {}).get('created_at', 'N/A')[:10] if invoices else 'N/A'}
+"""
+
+ def handle_amount_query(self, data: dict) -> str:
+ """Handle amount-related queries"""
+ invoices = data.get("invoices", [])
+ amounts = [inv.get('amount', 0) for inv in invoices if inv.get('amount', 0) > 0]
+
+ if not amounts:
+ return "No amount information found in invoices."
+
+ total_amount = sum(amounts)
+ avg_amount = total_amount / len(amounts)
+ max_amount = max(amounts)
+ min_amount = min(amounts)
+
+ # Find high-value invoices
+ high_value_threshold = sorted(amounts, reverse=True)[min(4, len(amounts)-1)] if len(amounts) > 5 else max_amount
+ high_value_invoices = [inv for inv in invoices if inv.get('amount', 0) >= high_value_threshold]
+
+ response = f"""
+**๐ฐ Financial Analysis**
+
+โข **Total Amount**: โน{total_amount:,.2f}
+โข **Average Amount**: โน{avg_amount:,.2f}
+โข **Highest Invoice**: โน{max_amount:,.2f}
+โข **Lowest Invoice**: โน{min_amount:,.2f}
+
+**๐ฏ High-Value Invoices (โน{high_value_threshold:,.2f}+)**
+"""
+
+ for i, inv in enumerate(high_value_invoices[:5], 1):
+ response += f"\n{i}. **{inv.get('invoice_number', 'N/A')}** - {inv.get('supplier_name', 'Unknown')} (โน{inv.get('amount', 0):,.2f})"
+
+ return response
+
+ def handle_supplier_query(self, data: dict, query: str) -> str:
+ """Handle supplier-related queries"""
+ invoices = data.get("invoices", [])
+
+ # Count invoices by supplier
+ supplier_counts = {}
+ supplier_amounts = {}
+
+ for inv in invoices:
+ supplier = inv.get('supplier_name', '').strip()
+ if supplier:
+ supplier_counts[supplier] = supplier_counts.get(supplier, 0) + 1
+ supplier_amounts[supplier] = supplier_amounts.get(supplier, 0) + inv.get('amount', 0)
+
+ if not supplier_counts:
+ return "No supplier information found in invoices."
+
+ # Sort suppliers by amount
+ top_suppliers = sorted(supplier_amounts.items(), key=lambda x: x[1], reverse=True)[:10]
+
+ response = f"""
+**๐ข Supplier Analysis**
+
+โข **Total Unique Suppliers**: {len(supplier_counts)}
+โข **Most Active**: {max(supplier_counts, key=supplier_counts.get)} ({supplier_counts[max(supplier_counts, key=supplier_counts.get)]} invoices)
+
+**๐ฐ Top Suppliers by Amount**
+"""
+
+ for i, (supplier, amount) in enumerate(top_suppliers, 1):
+ count = supplier_counts[supplier]
+ avg = amount / count if count > 0 else 0
+ response += f"\n{i}. **{supplier}** - โน{amount:,.2f} ({count} invoices, avg: โน{avg:,.2f})"
+
+ return response
+
+ def handle_semantic_search(self, query: str) -> str:
+ """Handle semantic search queries"""
+ try:
+ results = self.processor.vector_store.semantic_search(query, top_k=5)
+
+ if not results:
+ return f"No relevant results found for '{query}'. Try different keywords."
+
+ response = f"๐ **Semantic Search Results for '{query}'**\n\n"
+
+ for i, result in enumerate(results, 1):
+ response += f"{i}. **{result.invoice_number}** - {result.supplier_name}\n"
+ response += f" โข Similarity: {result.similarity_score:.3f}\n"
+ response += f" โข Amount: โน{result.metadata.get('amount', 0):,.2f}\n"
+ response += f" โข Preview: {result.content_preview[:100]}...\n\n"
+
+ return response
+
+ except Exception as e:
+ return f"Semantic search error: {e}"
+
+ def handle_general_query(self, data: dict, query: str) -> str:
+ """Handle general queries with keyword search"""
+ invoices = data.get("invoices", [])
+ query_words = query.lower().split()
+
+ # Simple keyword matching
+ matching_invoices = []
+ for inv in invoices:
+ text_to_search = (
+ inv.get('supplier_name', '') + ' ' +
+ inv.get('buyer_name', '') + ' ' +
+ inv.get('product_description', '') + ' ' +
+ inv.get('extraction_info', {}).get('raw_text_preview', '')
+ ).lower()
+
+ if any(word in text_to_search for word in query_words):
+ matching_invoices.append(inv)
+
+ if not matching_invoices:
+ return f"No invoices found matching '{query}'. Try different keywords or check the summary."
+
+ response = f"๐ **Found {len(matching_invoices)} invoices matching '{query}'**\n\n"
+
+ for i, inv in enumerate(matching_invoices[:5], 1):
+ response += f"{i}. **{inv.get('invoice_number', 'N/A')}** - {inv.get('supplier_name', 'Unknown')}\n"
+ response += f" โข Amount: โน{inv.get('amount', 0):,.2f}\n"
+ response += f" โข Date: {inv.get('date', 'N/A')}\n\n"
+
+ if len(matching_invoices) > 5:
+ response += f"... and {len(matching_invoices) - 5} more results."
+
+ return response
+
+# ===============================================================================
+# STREAMLIT APPLICATION
+# ===============================================================================
+
+def create_app():
+ """Main Streamlit application"""
+
+ # Generate unique session ID for this run
+ if 'session_id' not in st.session_state:
+ st.session_state.session_id = str(uuid.uuid4())[:8]
+
+ session_id = st.session_state.session_id
+
+ # Custom CSS
+ st.markdown("""
+
+ """, unsafe_allow_html=True)
+
+ # Header
+ st.markdown('๐ AI Invoice Processing System
', unsafe_allow_html=True)
+ st.markdown("""
+
+
+ AI-Powered Document Processing โข Semantic Search โข Smart Analytics โข Hugging Face Spaces
+
+
+
๐ค AI Extraction
+
Advanced NLP models extract structured data automatically
+
+
๐ Smart Search
+
Semantic search finds invoices using natural language
+
+
๐ Analytics
+
Comprehensive insights and visualizations
+
+
๐ AI Invoice Processing System - Optimized for Hugging Face Spaces
+
Built with โค๏ธ using Streamlit, Transformers, and AI
+
๐ AI Invoice Processing System
', unsafe_allow_html=True)
+ st.markdown("""
+
+
+ AI-Powered Document Processing โข Semantic Search โข Smart Analytics โข Hugging Face Spaces
+
+
+
๐ค AI Extraction
+
Advanced NLP models extract structured data automatically
+
+
๐ Smart Search
+
Semantic search finds invoices using natural language
+
+
๐ Analytics
+
Comprehensive insights and visualizations
+
+
๐ AI Invoice Processing System - Optimized for Hugging Face Spaces
+
Built with โค๏ธ using Streamlit, Transformers, and AI
+
๐ AI Invoice Processing System
', unsafe_allow_html=True)
+ st.markdown("""
+
+
+ AI-Powered Document Processing โข Semantic Search โข Smart Analytics โข Hugging Face Spaces
+
+
+
๐ค AI Extraction
+
Advanced NLP models extract structured data automatically
+
+
๐ Smart Search
+
Semantic search finds invoices using natural language
+
+
๐ Analytics
+
Comprehensive insights and visualizations
+
+
๐ AI Invoice Processing System - Optimized for Hugging Face Spaces
+
Built with โค๏ธ using Streamlit, Transformers, and AI
+