first version of gmail invoice extractor

Dockers/gmail-invoice-extractor/main.py

@@ -0,0 +1,258 @@
+import os
+import re
+import base64
+import pdfplumber
+import logging
+from datetime import datetime, timedelta
+from typing import List, Tuple, Optional
+
+from google.oauth2.credentials import Credentials
+from google_auth_oauthlib.flow import InstalledAppFlow
+from googleapiclient.discovery import build
+from google.auth.transport.requests import Request
+
+from config import Config
+
+# Load configuration
+config = Config()
+
+# Configure logging
+logging.basicConfig(
+    level=getattr(logging, config.log_level),
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('invoice_extractor.log'),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger(__name__)
+
+# Gmail API scope for read-only with attachments
+SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
+
+# Folder to save invoices
+SAVE_FOLDER = config.save_folder
+os.makedirs(SAVE_FOLDER, exist_ok=True)
+
+
+def get_last_month_date_range():
+    """Get the date range for the last month in Gmail query format."""
+    today = datetime.now()
+
+    # Calculate first day of current month
+    first_day_current = today.replace(day=1)
+
+    # Calculate first day of last month
+    if first_day_current.month == 1:
+        first_day_last_month = first_day_current.replace(year=first_day_current.year - 1, month=12)
+    else:
+        first_day_last_month = first_day_current.replace(month=first_day_current.month - 1)
+
+    # Calculate last day of last month
+    last_day_last_month = first_day_current - timedelta(days=1)
+
+    # Format dates for Gmail query (YYYY/MM/DD format)
+    start_date = first_day_last_month.strftime('%Y/%m/%d')
+    end_date = last_day_last_month.strftime('%Y/%m/%d')
+
+    logger.info(f"Scanning emails from {start_date} to {end_date}")
+
+    return f"after:{start_date} before:{end_date}"
+
+
+def gmail_authenticate():
+    """Authenticate with Gmail API using OAuth2."""
+    creds = None
+    if os.path.exists("token.json"):
+        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
+
+    if not creds or not creds.valid:
+        if creds and creds.expired and creds.refresh_token:
+            try:
+                creds.refresh(Request())
+                logger.info("Credentials refreshed successfully")
+            except Exception as e:
+                logger.error(f"Failed to refresh credentials: {e}")
+                creds = None
+
+        if not creds:
+            try:
+                flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
+                creds = flow.run_local_server(port=0)
+                logger.info("New credentials obtained")
+            except Exception as e:
+                logger.error(f"Failed to obtain credentials: {e}")
+                raise
+
+        with open("token.json", "w") as token:
+            token.write(creds.to_json())
+        logger.info("Credentials saved to token.json")
+
+    return build("gmail", "v1", credentials=creds)
+
+
+def search_messages(service, base_query: str, max_results: int = None) -> List[dict]:
+    """Search for messages matching the query with date filtering."""
+    if max_results is None:
+        max_results = config.max_messages
+
+    # Add date filter for last month
+    date_filter = get_last_month_date_range()
+    full_query = f"{base_query} {date_filter}"
+
+    try:
+        results = service.users().messages().list(
+            userId="me",
+            q=full_query,
+            maxResults=max_results
+        ).execute()
+        messages = results.get("messages", [])
+        logger.info(f"Found {len(messages)} messages matching query: {full_query}")
+        return messages
+    except Exception as e:
+        logger.error(f"Failed to search messages: {e}")
+        return []
+
+
+def get_attachments(service, msg_id: str) -> List[Tuple[str, bytes]]:
+    """Extract PDF attachments from a message."""
+    try:
+        message = service.users().messages().get(userId="me", id=msg_id).execute()
+        parts = message.get("payload", {}).get("parts", [])
+        attachments = []
+
+        for part in parts:
+            if part.get("filename") and part["filename"].lower().endswith(".pdf"):
+                att_id = part["body"].get("attachmentId")
+                if att_id:
+                    try:
+                        att = service.users().messages().attachments().get(
+                            userId="me", messageId=msg_id, id=att_id
+                        ).execute()
+                        data = base64.urlsafe_b64decode(att["data"])
+                        attachments.append((part["filename"], data))
+                        logger.debug(f"Extracted attachment: {part['filename']}")
+                    except Exception as e:
+                        logger.error(f"Failed to extract attachment {part['filename']}: {e}")
+
+        return attachments
+    except Exception as e:
+        logger.error(f"Failed to get attachments for message {msg_id}: {e}")
+        return []
+
+
+def extract_text_from_pdf(pdf_bytes: bytes) -> str:
+    """Extract text from PDF bytes."""
+    text = ""
+    temp_file = f"temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
+
+    try:
+        with open(temp_file, "wb") as f:
+            f.write(pdf_bytes)
+
+        with pdfplumber.open(temp_file) as pdf:
+            for page_num, page in enumerate(pdf.pages):
+                page_text = page.extract_text()
+                if page_text:
+                    text += page_text + "\n"
+                logger.debug(f"Extracted text from page {page_num + 1}")
+
+        return text
+    except Exception as e:
+        logger.error(f"Failed to extract text from PDF: {e}")
+        return ""
+    finally:
+        if os.path.exists(temp_file):
+            os.remove(temp_file)
+
+
+def contains_vat_number(text: str) -> bool:
+    """Check if text contains any of the configured VAT number patterns."""
+    if not text:
+        return False
+
+    text_upper = text.upper()
+    vat_patterns = config.get_vat_patterns()
+
+    for pattern in vat_patterns:
+        if re.search(pattern, text_upper):
+            logger.info(f"Found VAT number match with pattern: {pattern}")
+            return True
+
+    return False
+
+
+def save_invoice(filename: str, pdf_bytes: bytes) -> bool:
+    """Save invoice PDF to the designated folder."""
+    try:
+        # Create filename with timestamp to avoid conflicts
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        name, ext = os.path.splitext(filename)
+        safe_filename = f"{name}_{timestamp}{ext}"
+        save_path = os.path.join(SAVE_FOLDER, safe_filename)
+
+        with open(save_path, "wb") as f:
+            f.write(pdf_bytes)
+
+        logger.info(f"✅ Saved invoice: {safe_filename}")
+        return True
+    except Exception as e:
+        logger.error(f"Failed to save invoice {filename}: {e}")
+        return False
+
+
+def main():
+    """Main function to process Gmail messages and extract invoices."""
+    logger.info("Starting Gmail Invoice Extractor")
+    logger.info(f"Configuration: {config.gmail_query}, max_messages: {config.max_messages}")
+
+    try:
+        # Authenticate with Gmail
+        service = gmail_authenticate()
+
+        # Search for messages with PDF attachments from last month
+        messages = search_messages(service, config.gmail_query)
+
+        if not messages:
+            logger.info("No messages found with PDF attachments from last month")
+            return
+
+        processed_count = 0
+        saved_count = 0
+
+        for msg in messages:
+            try:
+                msg_id = msg["id"]
+                attachments = get_attachments(service, msg_id)
+
+                for filename, pdf_bytes in attachments:
+                    processed_count += 1
+                    logger.info(f"Processing attachment: {filename}")
+
+                    # Extract text from PDF
+                    text = extract_text_from_pdf(pdf_bytes)
+
+                    if not text:
+                        logger.warning(f"No text extracted from {filename}")
+                        continue
+
+                    # Check for VAT number
+                    if contains_vat_number(text):
+                        if save_invoice(filename, pdf_bytes):
+                            saved_count += 1
+                    else:
+                        logger.info(f"❌ Skipped (no VAT match): {filename}")
+
+            except Exception as e:
+                logger.error(f"Error processing message {msg.get('id', 'unknown')}: {e}")
+                continue
+
+        logger.info(f"Processing complete. Processed: {processed_count}, Saved: {saved_count}")
+
+    except Exception as e:
+        logger.error(f"Fatal error in main: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    main()