loreal-deliverables-airtable/airtable_connect.py

#!/usr/bin/env python3
"""
Airtable Connection Script
Connects to Airtable and searches for records containing "OMG JN 1737841"
"""

from pyairtable import Api
import json
from datetime import datetime, timedelta

# Configuration
PAT_TOKEN = "pat32GcLDAyyT4V45.928c9f011206e1b8d26608436f8dcef6feb3eabcfe59a27e70459ac6dff54dde"
BASE_ID = "appWy60RN6TeHDXx6"
TABLE_ID = "tblCXjVtaHnLTCulY"  # Using table ID instead of name
SEARCH_TERM = "API TEST RECORD"  # Searching for test records

def add_test_record():
    """Add a test record to the MASTER table with all fields populated"""
    try:
        print("Connecting to Airtable to add test record...")

        # Initialize API connection
        api = Api(PAT_TOKEN)
        table = api.table(BASE_ID, TABLE_ID)

        # Get current date and future dates for testing
        today = datetime.now().strftime("%Y-%m-%d")
        future_date = (datetime.now() + timedelta(days=30)).strftime("%Y-%m-%d")

        # Create test record with key fields (using existing dropdown values where required)
        test_record = {
            "PROJECT Number": 9999999,
            "PULL / PUSH": "PUSH",
            "HUB": "UKI",
            "DIVISION": "LDB",
            "BRAND": ["VICHY"],
            "COUNTRY": "PORTUGAL",
            "PRODUCT": ["Multi "],
            "PROJECT NAME": "*** API TEST RECORD - DO NOT USE FOR PRODUCTION ***",
            "APPROVER": "*** API TEST APPROVER ***",
            "RISK": "COMPLETE",
            "COMMENTS": "*** THIS IS A TEST RECORD CREATED VIA API *** - Created on " + today,
            "BOOKING DDL": future_date,

            # Translation fields
            "T.SALSIFY COPY STATUS": "UPLOADED TO SALSIFY",
            "T.DDL": future_date,
            "T.ASSET COPY STATUS": "ASSET COPY APPROVED",
            "T.DDL 2": future_date,
            "T.COMMENTS": "*** API TEST *** Translation comments",
            "T.BOX": "https://test.box.com/folder/api-test-translation",

            # Production fields
            "P.STATUS": "BRIEF SUBMITTED",
            "P.OPERA DDL": future_date,
            "P.COMMENTS": "*** API TEST *** Production comments - This is a test record",

            # Syndication fields
            "S.ASIN STATUS": "N/A",
            "S.AMAZON STATUS": "N/A",
            "S.RETAILER STATUS": "PLANNED",
            "S.OG DDL": future_date,
            "S.EXP. DDL": future_date,
            "S.START": future_date,
            "S.COMMENTS": "*** API TEST *** Syndication comments",

            # Links
            "OPERA Link": "https://test.opera.com/api-test-record",
            "SALSIFY Link": "https://test.salsify.com/api-test-record",
            "BOX Link": "https://test.box.com/api-test-record",

            # Other fields
            "EAN": True,
            "Supply Date": future_date,
            "GO LIVE": future_date,
            "CM": "*** API TEST CM ***",
            "PM": ["Lisa"]  # Using existing PM value
            # Removed DESIGNER as it's a dropdown field
        }

        # Create the record
        print("\nCreating test record...")
        created_record = table.create(test_record)

        print("\n" + "="*80)
        print("✓ TEST RECORD SUCCESSFULLY CREATED!")
        print("="*80)
        print(f"\nRecord ID: {created_record['id']}")
        print(f"Created Time: {created_record.get('createdTime', 'N/A')}")
        print("\nFields:")
        for field_name, field_value in created_record['fields'].items():
            print(f"  {field_name}: {field_value}")
        print("="*80)

        return created_record

    except Exception as e:
        print(f"Error creating test record: {str(e)}")
        raise


def connect_and_search():
    """Connect to Airtable and search for records"""
    try:
        print("Connecting to Airtable...")
        print(f"Base ID: {BASE_ID}")
        print(f"Table ID: {TABLE_ID}")
        print()

        # Initialize API connection
        api = Api(PAT_TOKEN)

        # Access the table
        table = api.table(BASE_ID, TABLE_ID)

        print(f"Connection initialized for table: {TABLE_ID}")
        print(f"\nSearching for records containing: '{SEARCH_TERM}'...\n")

        # Fetch all records
        all_records = table.all()
        print(f"Total records in table: {len(all_records)}\n")


        # Search for matching records
        matching_records = []
        for record in all_records:
            # Search across all fields
            record_str = json.dumps(record['fields']).lower()
            if SEARCH_TERM.lower() in record_str:
                matching_records.append(record)

        # Display results
        print(f"\nSearching for: '{SEARCH_TERM}'")
        if matching_records:
            print(f"Found {len(matching_records)} matching record(s):\n")
            print("=" * 80)

            for idx, record in enumerate(matching_records, 1):
                print(f"\nMatching Record {idx}:")
                print(f"Record ID: {record['id']}")
                print(f"Created Time: {record.get('createdTime', 'N/A')}")
                print("\nFields:")
                for field_name, field_value in record['fields'].items():
                    print(f"  {field_name}: {field_value}")
                print("-" * 80)
        else:
            print(f"No records found containing '{SEARCH_TERM}'")

        return matching_records

    except Exception as e:
        print(f"Error: {str(e)}")
        raise

if __name__ == "__main__":
    # Add a test record
    test_record = add_test_record()

    print("\n\n" + "="*80)
    print("Now searching for all API test records...")
    print("="*80 + "\n")

    # Search for test records
    results = connect_and_search()