#!/usr/bin/env python3
"""Seed the database with campaign templates from YAML files.
Loads all .yaml files from app/services/lead_gen/templates/ and creates
AdTemplate records. Idempotent ā skips templates that already exist by slug.
Usage:
python scripts/seed_templates.py # uses TEST_DATABASE_URI or DATABASE_URI
python scripts/seed_templates.py --dry # preview without writing
"""
import argparse
import glob
import os
import sys
import uuid
import yaml
# Ensure app package is importable
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Set test defaults (override with env vars for production)
os.environ.setdefault('SECRET_KEY', 'seed-template-dev-key')
os.environ.setdefault('DATABASE_URL', 'sqlite:///seed_templates_local.db')
from app import create_app
from app.models import db, AdTemplate
def load_yaml_templates(template_dir: str) -> list[dict]:
"""Load all .yaml/.yml files from template_dir and parse them."""
templates = []
patterns = [
os.path.join(template_dir, "**", "*.yaml"),
os.path.join(template_dir, "**", "*.yml"),
]
files = []
for pattern in patterns:
files.extend(glob.glob(pattern, recursive=True))
for filepath in sorted(files):
with open(filepath, "r") as f:
data = yaml.safe_load(f)
if not data or "name" not in data:
print(f" ā Skipping {filepath}: missing 'name' field")
continue
templates.append(data)
return templates
def yaml_to_ad_template(template_data: dict) -> dict:
"""Convert a YAML template dict to AdTemplate kwargs."""
name = template_data.get("name", "Untitled")
vertical = template_data.get("vertical", "other").lower()
# Build slug from name
slug = name.lower()
slug = slug.replace(" ", "-")
slug = "".join(c for c in slug if c.isalnum() or c == "-")
slug = f"{vertical}-{slug}"
# Extract budget recommendation
budget = None
budget_raw = template_data.get("budget_recommendation")
if budget_raw and isinstance(budget_raw, str):
# Parse "$X/month" or "$X" formats
cleaned = budget_raw.replace("$", "").replace("/", "").strip()
try:
budget = float(cleaned.split()[0])
except (ValueError, IndexError):
pass
elif isinstance(budget_raw, (int, float)):
budget = float(budget_raw)
# Extract target CPA
target_cpa = None
cpa_raw = template_data.get("target_cpa")
if cpa_raw and isinstance(cpa_raw, str):
cleaned = cpa_raw.replace("$", "").strip()
try:
target_cpa = float(cleaned)
except ValueError:
pass
elif isinstance(cpa_raw, (int, float)):
target_cpa = float(cpa_raw)
# Bidding strategy
bidding = template_data.get("bidding_strategy", "target_cpa")
# Convert structure to JSON-serializable dict
structure = {}
if "ad_groups" in template_data:
structure["ad_groups"] = []
for ag in template_data["ad_groups"]:
group = {
"name": ag.get("name", ""),
"keywords": [],
"ads": [],
}
# Keywords
for kw in ag.get("keywords", []):
if isinstance(kw, dict):
group["keywords"].append({
"text": kw.get("text", ""),
"match_type": kw.get("match_type", "PHRASE").upper(),
"max_cpc": kw.get("max_cpc"),
})
else:
group["keywords"].append({"text": str(kw), "match_type": "PHRASE"})
# Ads
for ad in ag.get("ads", []):
if isinstance(ad, dict):
group["ads"].append({
"headline1": ad.get("headline1", ""),
"headline2": ad.get("headline2", ad.get("headline", "")),
"headline3": ad.get("headline3", ""),
"description1": ad.get("description1", ad.get("description", "")),
"description2": ad.get("description2", ""),
"display_url": ad.get("display_url", ""),
"final_url": ad.get("final_url", ""),
})
else:
group["ads"].append({"headline1": str(ad)})
# Campaign settings
if "campaign" in template_data:
structure["campaign"] = {
"name_template": template_data["campaign"].get("name", ""),
"networks": template_data["campaign"].get("networks", ["search"]),
"locations": template_data["campaign"].get("locations", []),
"languages": template_data["campaign"].get("languages", ["en"]),
"ad_scheduling": template_data["campaign"].get("ad_scheduling", {}),
"audiences": template_data["campaign"].get("audiences", []),
"extensions": template_data["campaign"].get("extensions", {}),
}
# Platform-specific settings
platform = template_data.get("platform")
if platform:
structure["platform"] = platform
return {
"slug": slug,
"name": name,
"vertical": vertical,
"description": template_data.get("description", ""),
"budget_recommendation": budget,
"target_cpa": target_cpa,
"bidding_strategy": bidding,
"structure_json": structure,
"is_active": True,
}
def seed_templates(dry_run: bool = False) -> dict:
"""Load YAML templates and create/update AdTemplate records."""
app = create_app()
template_dir = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"app", "services", "lead_gen", "templates",
)
if not os.path.isdir(template_dir):
print(f"Template directory not found: {template_dir}")
return {"error": "template_dir_not_found", "path": template_dir}
yaml_templates = load_yaml_templates(template_dir)
if not yaml_templates:
print("No YAML templates found.")
return {"error": "no_templates_found"}
print(f"\nFound {len(yaml_templates)} YAML template(s):\n")
with app.app_context():
# Ensure tables exist
db.create_all()
created = 0
skipped = 0
errors = 0
for data in yaml_templates:
kwargs = yaml_to_ad_template(data)
slug = kwargs["slug"]
if dry_run:
created += 1
print(f" + {slug} ā would create")
continue
existing = AdTemplate.query.filter_by(slug=slug).first()
if existing:
skipped += 1
print(f" ā {slug} ā already exists (skipped)")
continue
try:
template = AdTemplate(**kwargs)
db.session.add(template)
db.session.flush()
created += 1
print(f" + {slug} ā created")
except Exception as e:
errors += 1
print(f" ā {slug} ā error: {e}")
db.session.rollback()
if not dry_run:
db.session.commit()
return {
"created": created,
"skipped": skipped,
"errors": errors,
"dry_run": dry_run,
}
def main():
parser = argparse.ArgumentParser(description="Seed ad campaign templates")
parser.add_argument("--dry", action="store_true", help="Preview without writing")
args = parser.parse_args()
print("=" * 60)
print(" Lead Gen Template Seeder")
print("=" * 60)
result = seed_templates(dry_run=args.dry)
if "error" in result:
print(f"\nā Failed: {result['error']}")
sys.exit(1)
print(f"\n{'[DRY RUN] ' if result['dry_run'] else ''}Results:")
print(f" Created: {result['created']}")
print(f" Skipped: {result['skipped']}")
print(f" Errors: {result['errors']}")
if result['errors'] > 0:
sys.exit(1)
if __name__ == "__main__":
main()