undefined

undefined

undefined

undefined

undefined

To get a Harvard Art Museums API key:
1. Visit https://www.harvardartmuseums.org/collections/api
2. Request an API key (free)
3. Add it to your `.env` file

undefined

undefined

-- Artifact metadata table
CREATE TABLE artifactmetadata (
    id INT PRIMARY KEY,
    title VARCHAR(500),
    culture VARCHAR(255),
    century VARCHAR(255),
    dated VARCHAR(255),
    classification VARCHAR(255),
    department VARCHAR(255),
    technique VARCHAR(500),
    medium VARCHAR(500),
    dimensions VARCHAR(500),
    creditline TEXT,
    accession_number VARCHAR(255),
    url TEXT
);

-- Artifact media table
CREATE TABLE artifactmedia (
    media_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    image_url TEXT,
    media_type VARCHAR(100),
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

-- Artifact colors table
CREATE TABLE artifactcolors (
    color_id INT AUTO_INCREMENT PRIMARY KEY,
    artifact_id INT,
    color VARCHAR(50),
    percentage FLOAT,
    FOREIGN KEY (artifact_id) REFERENCES artifactmetadata(id)
);

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def fetch_artifacts(api_key, page=1, size=100):
    """
    Fetch artifacts from Harvard Art Museums API
    
    Args:
        api_key: Harvard API key
        page: Page number for pagination
        size: Number of records per page (max 100)
    
    Returns:
        JSON response with artifact data
    """
    base_url = "https://api.harvardartmuseums.org/object"
    params = {
        'apikey': api_key,
        'page': page,
        'size': size,
        'hasimage': 1  # Only artifacts with images
    }
    
    response = requests.get(base_url, params=params)
    response.raise_for_status()
    
    return response.json()

undefined

import pandas as pd

def transform_artifacts(raw_data):
    """
    Transform raw API data into structured DataFrames
    
    Args:
        raw_data: JSON response from API
    
    Returns:
        Dictionary of DataFrames (metadata, media, colors)
    """
    artifacts = []
    media_list = []
    colors_list = []
    
    for record in raw_data['records']:
        # Extract metadata
        artifact = {
            'id': record.get('id'),
            'title': record.get('title', 'Unknown'),
            'culture': record.get('culture'),
            'century': record.get('century'),
            'dated': record.get('dated'),
            'classification': record.get('classification'),
            'department': record.get('department'),
            'technique': record.get('technique'),
            'medium': record.get('medium'),
            'dimensions': record.get('dimensions'),
            'creditline': record.get('creditline'),
            'accession_number': record.get('accessionnumber'),
            'url': record.get('url')
        }
        artifacts.append(artifact)
        
        # Extract media
        if 'images' in record and record['images']:
            for img in record['images']:
                media_list.append({
                    'artifact_id': record['id'],
                    'image_url': img.get('baseimageurl'),
                    'media_type': 'image'
                })
        
        # Extract colors
        if 'colors' in record and record['colors']:
            for color in record['colors']:
                colors_list.append({
                    'artifact_id': record['id'],
                    'color': color.get('color'),
                    'percentage': color.get('percent')
                })
    
    return {
        'metadata': pd.DataFrame(artifacts),
        'media': pd.DataFrame(media_list),
        'colors': pd.DataFrame(colors_list)
    }

import mysql.connector
from mysql.connector import Error

def get_db_connection():
    """Create database connection"""
    try:
        connection = mysql.connector.connect(
            host=os.getenv('DB_HOST'),
            port=os.getenv('DB_PORT', 3306),
            user=os.getenv('DB_USER'),
            password=os.getenv('DB_PASSWORD'),
            database=os.getenv('DB_NAME')
        )
        return connection
    except Error as e:
        print(f"Error connecting to database: {e}")
        return None

def load_artifacts_to_db(dataframes):
    """
    Load transformed data into SQL database
    
    Args:
        dataframes: Dictionary of DataFrames from transform_artifacts()
    """
    connection = get_db_connection()
    if not connection:
        return
    
    cursor = connection.cursor()
    
    try:
        # Insert metadata
        for _, row in dataframes['metadata'].iterrows():
            query = """
            INSERT INTO artifactmetadata 
            (id, title, culture, century, dated, classification, 
             department, technique, medium, dimensions, creditline, 
             accession_number, url)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE title=VALUES(title)
            """
            cursor.execute(query, tuple(row))
        
        # Insert media
        for _, row in dataframes['media'].iterrows():
            query = """
            INSERT INTO artifactmedia (artifact_id, image_url, media_type)
            VALUES (%s, %s, %s)
            """
            cursor.execute(query, tuple(row))
        
        # Insert colors
        for _, row in dataframes['colors'].iterrows():
            query = """
            INSERT INTO artifactcolors (artifact_id, color, percentage)
            VALUES (%s, %s, %s)
            """
            cursor.execute(query, tuple(row))
        
        connection.commit()
        print("Data loaded successfully!")
        
    except Error as e:
        print(f"Error loading data: {e}")
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

import streamlit as st
import plotly.express as px

def main():
    st.title("Harvard Art Museums Analytics Dashboard")
    
    # Sidebar configuration
    st.sidebar.header("Configuration")
    api_key = st.sidebar.text_input("API Key", type="password", 
                                     value=os.getenv('HARVARD_API_KEY', ''))
    
    num_records = st.sidebar.slider("Records to fetch", 10, 500, 100)
    
    # ETL Pipeline Section
    st.header("🔄 ETL Pipeline")
    
    if st.button("Run ETL Pipeline"):
        with st.spinner("Fetching data from API..."):
            raw_data = fetch_artifacts(api_key, page=1, size=num_records)
            st.success(f"Fetched {len(raw_data['records'])} artifacts")
        
        with st.spinner("Transforming data..."):
            transformed = transform_artifacts(raw_data)
            st.success("Data transformed successfully")
        
        with st.spinner("Loading to database..."):
            load_artifacts_to_db(transformed)
            st.success("Data loaded to database")
    
    # Analytics Section
    st.header("📊 Analytics")
    
    query_options = {
        "Artifacts by Culture": """
            SELECT culture, COUNT(*) as count 
            FROM artifactmetadata 
            WHERE culture IS NOT NULL 
            GROUP BY culture 
            ORDER BY count DESC 
            LIMIT 10
        """,
        "Artifacts by Century": """
            SELECT century, COUNT(*) as count 
            FROM artifactmetadata 
            WHERE century IS NOT NULL 
            GROUP BY century 
            ORDER BY count DESC
        """,
        "Top Colors Used": """
            SELECT color, SUM(percentage) as total_percentage 
            FROM artifactcolors 
            GROUP BY color 
            ORDER BY total_percentage DESC 
            LIMIT 10
        """
    }
    
    selected_query = st.selectbox("Select Analysis", list(query_options.keys()))
    
    if st.button("Run Query"):
        connection = get_db_connection()
        if connection:
            df = pd.read_sql(query_options[selected_query], connection)
            
            st.dataframe(df)
            
            # Visualization
            if len(df) > 0:
                fig = px.bar(df, x=df.columns[0], y=df.columns[1],
                            title=selected_query)
                st.plotly_chart(fig)
            
            connection.close()

if __name__ == "__main__":
    main()

-- 1. Distribution of artifacts by department
SELECT department, COUNT(*) as artifact_count
FROM artifactmetadata
GROUP BY department
ORDER BY artifact_count DESC;

-- 2. Artifacts with media vs without
SELECT 
    CASE WHEN m.artifact_id IS NOT NULL THEN 'Has Media' ELSE 'No Media' END as media_status,
    COUNT(*) as count
FROM artifactmetadata a
LEFT JOIN artifactmedia m ON a.id = m.artifact_id
GROUP BY media_status;

-- 3. Most common artifact classifications
SELECT classification, COUNT(*) as count
FROM artifactmetadata
WHERE classification IS NOT NULL
GROUP BY classification
ORDER BY count DESC
LIMIT 15;

-- 4. Color distribution across all artifacts
SELECT color, ROUND(AVG(percentage), 2) as avg_percentage, COUNT(*) as occurrences
FROM artifactcolors
GROUP BY color
ORDER BY occurrences DESC
LIMIT 20;

-- 5. Artifacts by culture and century
SELECT culture, century, COUNT(*) as count
FROM artifactmetadata
WHERE culture IS NOT NULL AND century IS NOT NULL
GROUP BY culture, century
ORDER BY count DESC
LIMIT 20;

import time

def fetch_with_rate_limit(api_key, total_records=500, batch_size=100):
    """Fetch data with rate limiting"""
    all_data = []
    pages = (total_records // batch_size) + 1
    
    for page in range(1, pages + 1):
        data = fetch_artifacts(api_key, page=page, size=batch_size)
        all_data.extend(data['records'])
        
        # Rate limit: 1 request per second
        time.sleep(1)
    
    return all_data

def verify_db_connection():
    """Test database connectivity"""
    try:
        conn = get_db_connection()
        if conn and conn.is_connected():
            print("✓ Database connection successful")
            cursor = conn.cursor()
            cursor.execute("SELECT DATABASE()")
            db_name = cursor.fetchone()[0]
            print(f"✓ Connected to database: {db_name}")
            cursor.close()
            conn.close()
            return True
    except Error as e:
        print(f"✗ Connection failed: {e}")
        return False

def safe_extract(record, key, default='Unknown'):
    """Safely extract nested data from API response"""
    try:
        value = record.get(key, default)
        return value if value else default
    except:
        return default

undefined

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def etl_pipeline(api_key, num_records=100):
    """Complete ETL pipeline with logging"""
    try:
        logger.info(f"Starting ETL for {num_records} records")
        
        raw_data = fetch_artifacts(api_key, size=num_records)
        logger.info(f"Extracted {len(raw_data['records'])} records")
        
        transformed = transform_artifacts(raw_data)
        logger.info(f"Transformed into {len(transformed['metadata'])} artifacts")
        
        load_artifacts_to_db(transformed)
        logger.info("Data loaded successfully")
        
        return True
    except Exception as e:
        logger.error(f"ETL pipeline failed: {e}")
        return False