112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
import sys
|
||
import os
|
||
import requests
|
||
import json
|
||
from sqlalchemy import create_engine, select
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
# Add paths to use backend models directly for complex seeding (Matrix/Person)
|
||
sys.path.append(os.path.join(os.getcwd(), "company-explorer"))
|
||
from backend.database import Base, Company, Contact, Industry, JobRoleMapping, MarketingMatrix
|
||
|
||
# Database Connection (Direct SQL access is easier for seeding specific IDs)
|
||
DB_PATH = "sqlite:///companies_v3_fixed_2.db" # Local relative path
|
||
engine = create_engine(DB_PATH)
|
||
Session = sessionmaker(bind=engine)
|
||
session = Session()
|
||
|
||
def seed():
|
||
print("--- Company Explorer Test Data Seeder ---")
|
||
print("This script prepares the database for the SuperOffice Connector End-to-End Test.")
|
||
|
||
# 1. User Input
|
||
so_contact_id = input("Enter SuperOffice Contact ID (Company) [e.g. 123]: ").strip()
|
||
so_person_id = input("Enter SuperOffice Person ID [e.g. 456]: ").strip()
|
||
company_name = input("Enter Company Name [e.g. Test GmbH]: ").strip() or "Test GmbH"
|
||
person_role = "Geschäftsführer" # Fixed for test simplicity
|
||
industry_name = "Logistik" # Fixed for test simplicity
|
||
|
||
if not so_contact_id or not so_person_id:
|
||
print("Error: IDs are required!")
|
||
return
|
||
|
||
print(f"\nSeeding for Company '{company_name}' (ID: {so_contact_id}) and Person (ID: {so_person_id})...")
|
||
|
||
# 2. Check/Create Industry
|
||
industry = session.query(Industry).filter_by(name=industry_name).first()
|
||
if not industry:
|
||
industry = Industry(name=industry_name, description="Test Industry")
|
||
session.add(industry)
|
||
session.commit()
|
||
print(f"✅ Created Industry '{industry_name}'")
|
||
else:
|
||
print(f"ℹ️ Industry '{industry_name}' exists")
|
||
|
||
# 3. Check/Create Job Role
|
||
role_map = session.query(JobRoleMapping).filter_by(role=person_role).first()
|
||
if not role_map:
|
||
role_map = JobRoleMapping(pattern=person_role, role=person_role) # Simple mapping
|
||
session.add(role_map)
|
||
session.commit()
|
||
print(f"✅ Created Role Mapping '{person_role}'")
|
||
else:
|
||
print(f"ℹ️ Role Mapping '{person_role}' exists")
|
||
|
||
# 4. Check/Create Company
|
||
company = session.query(Company).filter_by(crm_id=str(so_contact_id)).first()
|
||
if not company:
|
||
company = Company(
|
||
name=company_name,
|
||
crm_id=str(so_contact_id),
|
||
industry_ai=industry_name, # Link to our test industry
|
||
status="ENRICHED"
|
||
)
|
||
session.add(company)
|
||
session.commit()
|
||
print(f"✅ Created Company '{company_name}' with CRM-ID {so_contact_id}")
|
||
else:
|
||
company.industry_ai = industry_name # Ensure correct industry for test
|
||
session.commit()
|
||
print(f"ℹ️ Company '{company_name}' exists (Updated Industry)")
|
||
|
||
# 5. Check/Create Person
|
||
person = session.query(Contact).filter_by(so_person_id=int(so_person_id)).first()
|
||
if not person:
|
||
person = Contact(
|
||
company_id=company.id,
|
||
first_name="Max",
|
||
last_name="Mustermann",
|
||
so_person_id=int(so_person_id),
|
||
so_contact_id=int(so_contact_id),
|
||
role=person_role
|
||
)
|
||
session.add(person)
|
||
session.commit()
|
||
print(f"✅ Created Person with SO-ID {so_person_id}")
|
||
else:
|
||
person.role = person_role # Ensure role match
|
||
session.commit()
|
||
print(f"ℹ️ Person with SO-ID {so_person_id} exists (Updated Role)")
|
||
|
||
# 6. Check/Create Matrix Entry
|
||
matrix = session.query(MarketingMatrix).filter_by(industry_id=industry.id, role_id=role_map.id).first()
|
||
if not matrix:
|
||
matrix = MarketingMatrix(
|
||
industry_id=industry.id,
|
||
role_id=role_map.id,
|
||
subject="Test Betreff: Optimierung für {{company_name}}",
|
||
intro="Hallo, dies ist ein generierter Test-Text aus dem Company Explorer.",
|
||
social_proof="Wir arbeiten bereits erfolgreich mit anderen Logistikern zusammen."
|
||
)
|
||
session.add(matrix)
|
||
session.commit()
|
||
print(f"✅ Created Matrix Entry for {industry_name} x {person_role}")
|
||
else:
|
||
print(f"ℹ️ Matrix Entry exists")
|
||
|
||
print("\n🎉 Seeding Complete! The Company Explorer is ready.")
|
||
print(f"You can now trigger the Webhook for Contact {so_contact_id} / Person {so_person_id}.")
|
||
|
||
if __name__ == "__main__":
|
||
seed()
|