การผสาน CRM: Salesforce และ HubSpot

from simple_salesforce import Salesforce

sf = Salesforce(
    username='agent@company.com',
    password='YOUR_SF_PASSWORD',
    security_token='YOUR_SF_TOKEN',
    domain='login'   # use 'test' for sandbox
)
print('Connected to Salesforce org:', sf.base_url)

def get_customer_by_email(sf, email: str) -> dict | None:
    query = (
        f"SELECT Id, FirstName, LastName, Email, AccountId, "
        f"Account.Name, Account.Type "
        f"FROM Contact "
        f"WHERE Email = '{email}' "
        f"LIMIT 1"
    )
    result = sf.query(query)
    if result['totalSize'] == 0:
        return None
    record = result['records'][0]
    return {
        'contact_id': record['Id'],
        'name': f"{record['FirstName']} {record['LastName']}",
        'account_id': record['AccountId'],
        'account_name': record['Account']['Name']
    }

if __name__ == '__main__':
    class FakeSF:
        def query(self, query):
            return {
                'totalSize': 1,
                'records': [{
                    'Id': '003ABC', 'FirstName': 'Jane', 'LastName': 'Doe',
                    'AccountId': '001XYZ',
                    'Account': {'Name': 'Acme Corp', 'Type': 'Customer'},
                }],
            }
    customer = get_customer_by_email(FakeSF(), 'jane@acme.com')
    print(f"Found customer: {customer['name']} at {customer['account_name']}")

def get_case_history(sf, contact_id: str, limit: int = 10) -> list:
    query = (
        f"SELECT Id, CaseNumber, Subject, Status, "
        f"CreatedDate, ClosedDate, Resolution__c "
        f"FROM Case "
        f"WHERE ContactId = '{contact_id}' "
        f"ORDER BY CreatedDate DESC "
        f"LIMIT {limit}"
    )
    result = sf.query(query)
    return [
        {
            'case_number': r['CaseNumber'],
            'subject': r['Subject'],
            'status': r['Status'],
            'created': r['CreatedDate'][:10],
            'resolution': r.get('Resolution__c')
        }
        for r in result['records']
    ]

if __name__ == '__main__':
    class FakeSF:
        def query(self, query):
            return {'records': [{
                'Id': '500A', 'CaseNumber': '00001234', 'Subject': 'Login issue',
                'Status': 'Closed', 'CreatedDate': '2026-07-01T10:00:00Z',
                'ClosedDate': '2026-07-02T09:00:00Z', 'Resolution__c': 'Reset password',
            }]}
    for case in get_case_history(FakeSF(), '003ABC'):
        print(f"Case {case['case_number']}: {case['subject']} ({case['status']})")

def create_case(sf, contact_id: str, account_id: str,
                subject: str, description: str, priority: str = 'Medium') -> str:
    result = sf.Case.create({
        'ContactId': contact_id,
        'AccountId': account_id,
        'Subject': subject,
        'Description': description,
        'Priority': priority,
        'Status': 'New',
        'Origin': 'AI Agent'
    })
    case_id = result['id']
    print(f'Created Case: {case_id}')
    return case_id

if __name__ == '__main__':
    class FakeCase:
        def create(self, data):
            return {'id': '500NEW01'}
    class FakeSF:
        Case = FakeCase()

    create_case(FakeSF(), '003ABC', '001XYZ', 'Cannot reset password', 'Customer locked out after 3 attempts')

from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput

client = HubSpot(access_token='YOUR_HUBSPOT_PRIVATE_APP_TOKEN')
print('HubSpot client ready')

from hubspot.crm.contacts import PublicObjectSearchRequest

def get_hubspot_contact(client, email: str) -> dict | None:
    search_request = PublicObjectSearchRequest(
        filter_groups=[{
            'filters': [{
                'propertyName': 'email',
                'operator': 'EQ',
                'value': email
            }]
        }],
        properties=['firstname', 'lastname', 'email',
                    'lifecyclestage', 'hs_object_id']
    )
    response = client.crm.contacts.search_api.do_search(
        public_object_search_request=search_request
    )
    if response.total == 0:
        return None
    contact = response.results[0]
    return {'id': contact.id, **contact.properties}

def get_hubspot_tickets(client, contact_id: str) -> list:
    assoc = client.crm.contacts.associations_api.get_all(
        contact_id,
        to_object_type='tickets'
    )
    ticket_ids = [r.id for r in assoc.results]
    if not ticket_ids:
        return []

    tickets = []
    for tid in ticket_ids[:10]:  # limit to 10 most recent
        ticket = client.crm.tickets.basic_api.get_by_id(
            tid,
            properties=['subject', 'hs_ticket_priority',
                        'hs_pipeline_stage', 'createdate']
        )
        tickets.append({
            'id': tid,
            'subject': ticket.properties.get('subject'),
            'priority': ticket.properties.get('hs_ticket_priority'),
            'stage': ticket.properties.get('hs_pipeline_stage')
        })
    return tickets

if __name__ == '__main__':
    from types import SimpleNamespace

    class AssocApi:
        def get_all(self, contact_id, to_object_type):
            return SimpleNamespace(results=[SimpleNamespace(id='t1'), SimpleNamespace(id='t2')])

    class TicketsBasicApi:
        _data = {
            't1': {'subject': 'Cannot login', 'hs_ticket_priority': 'HIGH', 'hs_pipeline_stage': '1'},
            't2': {'subject': 'Refund request', 'hs_ticket_priority': 'MEDIUM', 'hs_pipeline_stage': '2'},
        }
        def get_by_id(self, tid, properties):
            return SimpleNamespace(properties=self._data[tid])

    fake_client = SimpleNamespace(crm=SimpleNamespace(
        contacts=SimpleNamespace(associations_api=AssocApi()),
        tickets=SimpleNamespace(basic_api=TicketsBasicApi()),
    ))

    for t in get_hubspot_tickets(fake_client, 'contact_1'):
        print(f"Ticket {t['id']}: {t['subject']} [{t['priority']}]")

from hubspot.crm.tickets import SimplePublicObjectInputForCreate
from hubspot.crm.associations.v4.models import AssociationSpec

def create_hubspot_ticket(client, contact_id: str,
                          subject: str, description: str) -> str:
    ticket_input = SimplePublicObjectInputForCreate(
        properties={
            'subject': subject,
            'content': description,
            'hs_ticket_priority': 'MEDIUM',
            'hs_pipeline': 'support_pipeline',
            'hs_pipeline_stage': 'new'
        },
        associations=[{
            'types': [AssociationSpec(association_category='HUBSPOT_DEFINED',
                                     association_type_id=16)],
            'to': {'id': contact_id}
        }]
    )
    ticket = client.crm.tickets.basic_api.create(
        simple_public_object_input_for_create=ticket_input
    )
    return ticket.id

def update_hubspot_contact(client, contact_id: str, updates: dict) -> None:
    client.crm.contacts.basic_api.update(
        contact_id=contact_id,
        simple_public_object_input=SimplePublicObjectInput(
            properties=updates
        )
    )

# After resolving a billing issue:
update_hubspot_contact(client, 'CONTACT_123', {
    'hs_lead_status': 'CONNECTED',
    'notes_last_contacted': 'Resolved billing duplicate charge issue via AI agent.'
})

def get_customer_context(email: str, crm: str = 'salesforce') -> dict:
    if crm == 'salesforce':
        customer = get_customer_by_email(sf, email)
        if not customer:
            return {'found': False}
        cases = get_case_history(sf, customer['contact_id'])
        return {'found': True, 'customer': customer, 'history': cases}
    elif crm == 'hubspot':
        customer = get_hubspot_contact(client, email)
        if not customer:
            return {'found': False}
        tickets = get_hubspot_tickets(client, customer['id'])
        return {'found': True, 'customer': customer, 'history': tickets}
    return {'found': False, 'error': 'Unknown CRM'}