Floke
7c82e4b5d7
This commit introduces a new unsubscribe feature to allow contacts to opt-out from marketing automation. Key changes include: - Database schema migration: Added (UUID) to the model. - Data population: Implemented a script to assign unique tokens to existing contacts. - API endpoint: Created a public GET endpoint to handle opt-out requests. - Automation: New contacts automatically receive an unsubscribe token upon creation. - Integration: The full unsubscribe link is now returned via the provisioning API for storage in SuperOffice UDFs (ProgID: SuperOffice:9). - Documentation: Updated and to reflect the new feature and its integration requirements. - Added for quick overview and next steps.
44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
import uuid
|
|
import os
|
|
import sys
|
|
|
|
# This is the crucial part to fix the import error.
|
|
# We add the 'company-explorer' directory to the path, so imports can be absolute
|
|
# from the 'backend' module.
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
|
|
|
from backend.database import Contact, SessionLocal
|
|
|
|
def migrate_existing_contacts():
|
|
"""
|
|
Generates and adds an unsubscribe_token for all existing contacts
|
|
that do not have one yet.
|
|
"""
|
|
db = SessionLocal()
|
|
try:
|
|
contacts_to_update = db.query(Contact).filter(Contact.unsubscribe_token == None).all()
|
|
|
|
if not contacts_to_update:
|
|
print("All contacts already have an unsubscribe token. No migration needed.")
|
|
return
|
|
|
|
print(f"Found {len(contacts_to_update)} contacts without an unsubscribe token. Generating tokens...")
|
|
|
|
for contact in contacts_to_update:
|
|
token = str(uuid.uuid4())
|
|
contact.unsubscribe_token = token
|
|
print(f" - Generated token for contact ID {contact.id} ({contact.email})")
|
|
|
|
db.commit()
|
|
print("\nSuccessfully updated all contacts with new unsubscribe tokens.")
|
|
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting migration: Populating unsubscribe_token for existing contacts.")
|
|
migrate_existing_contacts()
|
|
print("Migration finished.") |