DatabaseCreation: seed_generator.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "faker==40.38.0",
# ]
# ///
"""Generate synthetic vendor, client, project, review and dispute data as CSV files.

Usage:  uv run seed_generator.py [out_dir] [--scale FLOAT] [--seed INT]

The output is deterministic: every date is derived from REFERENCE_DATE instead of
the clock, and both random and Faker are seeded, so the same seed, scale and code
always produce the same files.
"""

import argparse
import csv
import hashlib
import math
import random
import re
import time
from array import array
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from pathlib import Path

from faker import Faker

# ── CONFIG ────────────────────────────────────────────────────────────────────
REFERENCE_DATE = date(2026, 5, 13)   # fixed "today" for the generated data, for reproducible dates
REFERENCE_END = datetime.combine(REFERENCE_DATE, datetime.max.time()).replace(microsecond=0)
HISTORY_DAYS = 4 * 365               # maximum lookback for generated start dates
PROGRESS_EVERY = 100_000

fake = Faker()


# ── TABLE SIZES ───────────────────────────────────────────────────────────────
# Target sizes at --scale 1.0; --scale multiplies each of them (minimum 1).
# The number of reviews is capped by the number of completed projects. The
# lookup tables are fixed lists and do not scale.
NUM_VENDORS = 5_000
NUM_CLIENTS = 20_000
NUM_CLIENT_USERS = 50_000
NUM_VENDOR_USERS = 25_000
NUM_MGMT_USERS = 2_000
NUM_CONTRACTS = 250_000
NUM_PROJECTS = 1_400_000
NUM_REVIEWS = 1_000_000

# Probability that a review dated before REFERENCE_DATE gets a dispute.
DISPUTE_RATE = 0.15


# ── HELPERS ───────────────────────────────────────────────────────────────────
def scaled(n: int, scale: float) -> int:
    return max(1, round(n * scale))


def positive_float(text: str) -> float:
    value = float(text)
    if value <= 0:
        raise argparse.ArgumentTypeError("must be greater than 0")
    return value


@contextmanager
def csv_file(out_dir: Path, filename: str, header: list[str]):
    with (out_dir / filename).open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(header)
        yield writer


def report(filename: str, rows: int, t0: float) -> None:
    print(f"  ✓  {filename:<48}  {rows:>12,} rows   ({time.time() - t0:.1f}s)")


def write_table(out_dir: Path, filename: str, header: list[str], rows: list) -> None:
    t0 = time.time()
    with csv_file(out_dir, filename, header) as writer:
        writer.writerows(rows)
    report(filename, len(rows), t0)


def day_between(start: date, end: date) -> date:
    return start + timedelta(days=random.randint(0, (end - start).days))


def ts_on(day: date) -> datetime:
    """A timestamp during office hours (08:00-18:00) on the given day."""
    return datetime.combine(day, datetime.min.time()) + timedelta(seconds=random.randint(8 * 3600, 18 * 3600))


def ts_between(start: datetime, end: datetime) -> datetime:
    return start + timedelta(seconds=random.randint(0, int((end - start).total_seconds())))


def ts_after(start: datetime, max_days: int) -> datetime:
    """A timestamp up to max_days after start, never past REFERENCE_END."""
    return ts_between(start, min(start + timedelta(days=max_days), REFERENCE_END))


def ts_strictly_between(start: datetime, end: datetime, count: int) -> list[datetime]:
    """count distinct timestamps in ascending order, strictly between start and end."""
    if count == 0:
        return []
    offsets = random.sample(range(1, int((end - start).total_seconds())), count)
    return [start + timedelta(seconds=s) for s in sorted(offsets)]


def money(value: float) -> str:
    return f"{value:.2f}"


def slug(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")


def password_hash() -> str:
    return hashlib.sha256(fake.password().encode()).hexdigest()


# ═════════════════════════════════════════════════════════════════════════════
# 1. LOOKUP TABLES
# ═════════════════════════════════════════════════════════════════════════════
ROLES = [
    (1, "Admin", "Full platform access"),
    (2, "Support", "Customer support operations"),
    (3, "Finance", "Financial oversight and reporting"),
    (4, "Operations", "Day-to-day operational management"),
    (5, "Compliance", "Auditing, legal, and compliance"),
]
ROLE_IDS = [r[0] for r in ROLES]
ROLE_WEIGHTS = [5, 40, 15, 25, 15]   # relative sampling weights, in ROLES order

PERMISSIONS = [
    "view_projects",
    "edit_projects",
    "delete_projects",
    "view_reviews",
    "publish_reviews",
    "delete_reviews",
    "manage_users",
    "view_financials",
    "manage_subscriptions",
    "resolve_disputes",
    "assign_tickets",
    "view_reports",
    "export_data",
    "manage_roles",
    "impersonate_user",
]

ROLE_PERMISSIONS = {
    "Admin": PERMISSIONS,
    "Support": [
        "view_projects", "view_reviews", "view_reports", "assign_tickets", "resolve_disputes",
    ],
    "Finance": [
        "view_projects", "view_financials", "view_reports", "export_data", "manage_subscriptions",
    ],
    "Operations": [
        "view_projects", "edit_projects", "delete_projects", "view_reviews", "publish_reviews",
        "assign_tickets", "view_reports", "manage_users",
    ],
    "Compliance": [
        "view_projects", "view_reviews", "delete_reviews", "view_financials", "view_reports",
        "export_data", "resolve_disputes",
    ],
}

INDUSTRIES = [
    "Technology",
    "Healthcare",
    "Finance",
    "Retail",
    "Manufacturing",
    "Education",
    "Real Estate",
    "Energy",
    "Transportation",
    "Media",
    "Hospitality",
    "Agriculture",
    "Construction",
    "Pharmaceuticals",
    "Telecommunications",
    "Automotive",
    "Aerospace",
    "Legal Services",
    "Non-Profit",
    "Government",
]

STATUSES = ["Draft", "In Progress", "On Hold", "Completed", "Cancelled", "Under Review"]
STATUS_ID = {name: i for i, name in enumerate(STATUSES, start=1)}
FINISHED = {"Completed", "Cancelled"}

# The status a project has reached by the reference day, and how likely each is.
# status_path builds the statuses that led there: a prefix of
# Draft -> Under Review -> In Progress, optional hold/resume cycles, and a
# cancellation from any intermediate stage.
OUTCOMES = ["Completed", "Cancelled", "In Progress", "On Hold", "Under Review", "Draft"]
OUTCOME_WEIGHTS = [72, 5, 13, 4, 3, 3]
PROJECT_DAYS = {"Completed": (30, 730), "Cancelled": (7, 365)}   # min/max duration of a finished project

STATUS_COMMENTS = {
    "Draft": [
        "Project created from the contract scope.",
        "Initial draft entered by the vendor.",
        "Scope drafted; awaiting client review.",
    ],
    "Under Review": [
        "Scope and budget sent to the client for approval.",
        "Under review by the client's steering committee.",
        "Awaiting sign-off on the statement of work.",
    ],
    "In Progress": [
        "Kick-off meeting held; work started.",
        "Approved by the client; development under way.",
        "Resumed after the hold was lifted.",
        "Sprint work in progress.",
    ],
    "On Hold": [
        "Paused at the client's request.",
        "Waiting for third-party API access.",
        "On hold until the budget revision is approved.",
    ],
    "Completed": [
        "All deliverables accepted by the client.",
        "Final release deployed and handed over.",
        "Closed after the acceptance test.",
    ],
    "Cancelled": [
        "Cancelled by the client before delivery.",
        "Cancelled: the requirements changed substantially.",
        "Contract terminated; project cancelled.",
    ],
}

TECHNOLOGIES = [
    "Python",
    "JavaScript",
    "TypeScript",
    "Java",
    "C#",
    "Go",
    "Rust",
    "PHP",
    "Ruby",
    "Swift",
    "Kotlin",
    "React",
    "Angular",
    "Vue.js",
    "Node.js",
    "Django",
    "FastAPI",
    "Spring Boot",
    "PostgreSQL",
    "MySQL",
    "MongoDB",
    "Redis",
    "Elasticsearch",
    "Docker",
    "Kubernetes",
    "AWS",
    "Azure",
    "GCP",
    "Terraform",
    "GraphQL",
]
TECH_IDS = list(range(1, len(TECHNOLOGIES) + 1))

# (dimension_id, name, description, bias). The bias shifts every vendor's mean
# score on that dimension; the synthetic biases favour communication over
# deadline adherence and documentation.
DIMENSIONS = [
    (1,  "Code Quality",         "Quality, readability and maintainability of the delivered code",     0.00),
    (2,  "Communication",        "Clarity and frequency of communication throughout the project",      0.20),
    (3,  "Deadline Adherence",   "Delivery of milestones and the final work on schedule",             -0.25),
    (4,  "Overall Satisfaction", "Overall satisfaction with the vendor and the delivered result",       0.10),
    (5,  "Budget Adherence",     "Accuracy to the originally quoted budget",                          -0.20),
    (6,  "Technical Skill",      "Depth and breadth of technical expertise demonstrated",              0.15),
    (7,  "Responsiveness",       "Speed and attentiveness when addressing client queries",             0.10),
    (8,  "Documentation",        "Quality and completeness of the technical documentation delivered", -0.30),
    (9,  "Problem Solving",      "Effectiveness at resolving blockers and unexpected challenges",      0.00),
    (10, "Team Collaboration",   "How well the vendor team collaborated with client stakeholders",     0.05),
]

TIERS = [
    (1, "Starter",      99.00, False),
    (2, "Professional", 299.00, False),
    (3, "Business",     None,  True),
    (4, "Enterprise",   None,  True),
]
TIER_IDS = [t[0] for t in TIERS]
CUSTOM_PRICE_RANGE = {3: (350, 1_500), 4: (1_500, 6_000)}
SUBSCRIPTION_DAYS = [30, 90, 180, 365]

CURRENCIES = ["USD", "EUR", "GBP", "CHF", "CAD", "AUD"]
CURRENCY_WEIGHTS = [40, 35, 10, 5, 5, 5]
TLDS = ["com", "io", "net", "co", "dev"]


def write_lookup_tables(out_dir: Path) -> None:
    write_table(out_dir, "role.csv", ["role_id", "role_name", "description"], ROLES)
    write_table(out_dir, "permission.csv", ["permission_id", "action_name"],
                list(enumerate(PERMISSIONS, start=1)))
    write_table(out_dir, "industry.csv", ["industry_id", "industry_name"],
                list(enumerate(INDUSTRIES, start=1)))
    write_table(out_dir, "project_status.csv", ["status_id", "status_name"],
                list(enumerate(STATUSES, start=1)))
    write_table(out_dir, "technology.csv", ["technology_id", "technology_name"],
                list(enumerate(TECHNOLOGIES, start=1)))
    write_table(out_dir, "rating_dimension.csv", ["dimension_id", "dimension_name", "description"],
                [d[:3] for d in DIMENSIONS])
    write_table(out_dir, "subscription_tier.csv", ["tier_id", "tier_name", "list_price", "allows_custom_pricing"],
                TIERS)
    permission_id = {name: i for i, name in enumerate(PERMISSIONS, start=1)}
    write_table(out_dir, "role_permission.csv", ["role_id", "permission_id"],
                [(role_id, permission_id[p]) for role_id, role_name, _ in ROLES for p in ROLE_PERMISSIONS[role_name]])


# ═════════════════════════════════════════════════════════════════════════════
# 2. VENDORS & CLIENTS
# ═════════════════════════════════════════════════════════════════════════════
def write_vendors(out_dir: Path, n_vendors: int) -> tuple[list[str], list[float]]:
    """Returns each vendor's e-mail domain and its latent quality: the mean score
    its reviews are drawn around."""
    t0 = time.time()
    domains, quality = [], []
    with csv_file(out_dir, "vendor.csv", ["vendor_id", "agency_name", "website"]) as writer:
        for vendor_id in range(1, n_vendors + 1):
            name = fake.company()
            domain = f"{slug(name)}.{random.choice(TLDS)}"
            writer.writerow([vendor_id, name, f"https://www.{domain}/"])
            domains.append(domain)
            quality.append(min(4.8, max(1.5, random.gauss(3.6, 0.6))))
    report("vendor.csv", n_vendors, t0)
    return domains, quality


def write_clients(out_dir: Path, n_clients: int) -> list[str]:
    """Returns each client's e-mail domain."""
    t0 = time.time()
    domains = []
    with csv_file(out_dir, "client.csv", ["client_id", "industry_id", "company_name", "contact_email"]) as writer:
        for client_id in range(1, n_clients + 1):
            name = fake.company()
            domain = f"{slug(name)}.{random.choice(TLDS)}"
            mailbox = random.choice(["info", "contact", "office"])
            writer.writerow([client_id, random.randint(1, len(INDUSTRIES)), name, f"{mailbox}@{domain}"])
            domains.append(domain)
    report("client.csv", n_clients, t0)
    return domains


# ═════════════════════════════════════════════════════════════════════════════
# 3. USERS
# ═════════════════════════════════════════════════════════════════════════════
USER_COLUMNS = [
    "user_id", "type", "first_name", "last_name", "email", "password_hash",
    "is_active", "last_login_at", "created_at", "updated_at",
]


def write_users(
    out_dir: Path,
    n_client_users: int,
    n_vendor_users: int,
    n_mgmt_users: int,
    client_domains: list[str],
    vendor_domains: list[str],
) -> tuple[list[list[int]], list[list[int]], list[int]]:
    """Writes user.csv and the three subtype files. Users are dealt round-robin over
    the companies, so with at least as many users as companies every client and
    every vendor gets a user. Returns the user ids per client, per vendor, and the
    management user ids."""
    t0 = time.time()
    earliest = datetime.combine(REFERENCE_DATE - timedelta(days=HISTORY_DAYS + 30), datetime.min.time())
    used_emails: set[str] = set()
    client_to_users: list[list[int]] = [[] for _ in client_domains]
    vendor_to_users: list[list[int]] = [[] for _ in vendor_domains]
    mgmt_user_ids: list[int] = []
    user_id = 0

    def user_row(user_type: str, domain: str) -> list:
        nonlocal user_id
        user_id += 1
        first, last = fake.first_name(), fake.last_name()
        local = slug(f"{first} {last}").replace("-", ".")
        email, n = f"{local}@{domain}", 1
        while email in used_emails:
            n += 1
            email = f"{local}{n}@{domain}"
        used_emails.add(email)
        created = ts_between(earliest, REFERENCE_END)
        # inactive seed users are modelled as unconfirmed, with no login history
        is_active = random.random() < 0.85
        last_login = ts_between(created, REFERENCE_END) if is_active and random.random() < 0.90 else None
        updated = ts_between(last_login or created, REFERENCE_END)
        return [user_id, user_type, first, last, email, password_hash(), is_active, last_login, created, updated]

    with (
        csv_file(out_dir, "user.csv", USER_COLUMNS) as users,
        csv_file(out_dir, "client_user.csv", ["user_id", "client_id"]) as client_users,
        csv_file(out_dir, "vendor_user.csv", ["user_id", "vendor_id"]) as vendor_users,
        csv_file(out_dir, "management_user.csv", ["user_id", "role_id"]) as mgmt_users,
    ):
        for i in range(n_client_users):
            client_id = i % len(client_domains) + 1
            row = user_row("client", client_domains[client_id - 1])
            users.writerow(row)
            client_users.writerow([row[0], client_id])
            client_to_users[client_id - 1].append(row[0])
        for i in range(n_vendor_users):
            vendor_id = i % len(vendor_domains) + 1
            row = user_row("vendor", vendor_domains[vendor_id - 1])
            users.writerow(row)
            vendor_users.writerow([row[0], vendor_id])
            vendor_to_users[vendor_id - 1].append(row[0])
        for _ in range(n_mgmt_users):
            row = user_row("management", "vqems.com")
            users.writerow(row)
            mgmt_users.writerow([row[0], random.choices(ROLE_IDS, ROLE_WEIGHTS)[0]])
            mgmt_user_ids.append(row[0])
    report("user.csv", user_id, t0)
    report("client_user.csv", n_client_users, t0)
    report("vendor_user.csv", n_vendor_users, t0)
    report("management_user.csv", n_mgmt_users, t0)
    return client_to_users, vendor_to_users, mgmt_user_ids


# ═════════════════════════════════════════════════════════════════════════════
# 4. VENDOR SUBSCRIPTIONS
# ═════════════════════════════════════════════════════════════════════════════
SUBSCRIPTION_COLUMNS = [
    "contract_id", "vendor_id", "tier_id", "negotiated_price", "start_date", "end_date",
    "is_active", "created_at", "updated_at",
]


def write_subscriptions(out_dir: Path, n_vendors: int) -> None:
    """Up to three non-overlapping periods per vendor; only the last period can be
    active, and only while it has not ended."""
    t0 = time.time()
    contract_id = 0
    with csv_file(out_dir, "vendor_subscription.csv", SUBSCRIPTION_COLUMNS) as writer:
        for vendor_id in range(1, n_vendors + 1):
            periods = []
            start = REFERENCE_DATE - timedelta(days=random.randint(1, HISTORY_DAYS))
            for _ in range(random.choices([0, 1, 2, 3], weights=[10, 45, 30, 15])[0]):
                tier_id = random.choice(TIER_IDS)
                price = money(random.uniform(*CUSTOM_PRICE_RANGE[tier_id])) if tier_id in CUSTOM_PRICE_RANGE else None
                end = None if random.random() < 0.4 else start + timedelta(days=random.choice(SUBSCRIPTION_DAYS))
                periods.append((tier_id, price, start, end, ts_on(start)))
                if end is None:
                    break                                            # an open-ended period is the last one
                start = end + timedelta(days=random.randint(1, 60))
                if start > REFERENCE_DATE:
                    break                                            # that renewal has not happened yet
            for k, (tier_id, price, start, end, created) in enumerate(periods):
                contract_id += 1
                is_last = k == len(periods) - 1
                is_active = is_last and (end is None or end >= REFERENCE_DATE)
                updated = created if is_last else periods[k + 1][4]   # deactivated when the next period was created
                writer.writerow([contract_id, vendor_id, tier_id, price, start, end, is_active, created, updated])
    report("vendor_subscription.csv", contract_id, t0)


# ═════════════════════════════════════════════════════════════════════════════
# 5. CONTRACTS & PROJECTS
# ═════════════════════════════════════════════════════════════════════════════
CONTRACT_COLUMNS = [
    "contract_id", "client_id", "vendor_id", "contract_number", "contract_title", "start_date", "end_date",
    "total_value", "currency_code", "terms_summary", "is_active", "created_at", "updated_at",
]
PROJECT_COLUMNS = [
    "project_id", "contract_id", "status_id", "project_name", "start_date", "end_date", "budget",
    "created_at", "updated_at",
]
AUDIT_COLUMNS = ["audit_id", "project_id", "old_budget", "new_budget", "created_at", "updated_at"]
HISTORY_COLUMNS = [
    "project_status_history_id", "project_id", "vendor_user_id", "management_user_id", "status_id",
    "changed_at", "comment",
]


def status_path(outcome: str) -> list[str]:
    """The ordered statuses a project went through to reach the given outcome."""
    if outcome == "Cancelled":
        stage = random.choice(["Draft", "Under Review", "In Progress", "On Hold"])
        return status_path(stage) + ["Cancelled"]
    path = ["Draft"]
    if outcome == "Draft":
        return path
    path.append("Under Review")
    if outcome == "Under Review":
        return path
    path.append("In Progress")
    for _ in range(random.choices([0, 1, 2], weights=[80, 15, 5])[0]):
        path += ["On Hold", "In Progress"]
    if outcome in ("On Hold", "Completed"):
        path.append(outcome)
    return path


def draw_budget() -> int:
    """A project budget between 5,000 and 500,000, log-uniform and rounded to hundreds."""
    return int(round(math.exp(random.uniform(math.log(5_000), math.log(500_000))), -2))


def revised_budget(budget: int) -> int:
    """A budget cut or raise, rounded to hundreds."""
    factor = random.uniform(0.7, 0.9) if random.random() < 0.4 else random.uniform(1.1, 1.6)
    return int(round(budget * factor, -2))


def write_contracts_and_projects(
    out_dir: Path,
    n_contracts: int,
    n_projects: int,
    n_clients: int,
    n_vendors: int,
    vendor_to_users: list[list[int]],
    mgmt_user_ids: list[int],
) -> tuple[array, array, tuple[array, array, array]]:
    """One pass writes the contracts, their projects and the project technology,
    status history and budget audit files, so that (by convention of this seed)
    projects lie inside their contract's window, the contract value covers its
    project budgets, and the status history and budget audit of a project are
    consistent with the project row. Returns the client and vendor of every
    contract and the completed projects (id, contract id, end date ordinal)."""
    t0 = time.time()
    # every contract gets one project; the remaining projects are spread over the
    # contracts at random
    counts = array("i", [1]) * n_contracts
    for _ in range(max(0, n_projects - n_contracts)):
        counts[random.randrange(n_contracts)] += 1

    contract_client, contract_vendor = array("i"), array("i")
    completed_pid, completed_cid, completed_end = array("i"), array("i"), array("i")
    project_id = audit_id = history_id = n_technologies = 0
    print(f"     dot = {PROGRESS_EVERY:,} projects:  ", end="", flush=True)
    with (
        csv_file(out_dir, "client_vendor_contract.csv", CONTRACT_COLUMNS) as contracts,
        csv_file(out_dir, "project.csv", PROJECT_COLUMNS) as projects,
        csv_file(out_dir, "project_technology.csv", ["project_id", "technology_id"]) as technologies,
        csv_file(out_dir, "project_budget_audit.csv", AUDIT_COLUMNS) as audits,
        csv_file(out_dir, "project_status_history.csv", HISTORY_COLUMNS) as history,
    ):
        for contract_id in range(1, n_contracts + 1):
            client_id = random.randint(1, n_clients)
            vendor_id = random.randint(1, n_vendors)
            vendor_users = vendor_to_users[vendor_id - 1]
            c_start = REFERENCE_DATE - timedelta(days=random.randint(1, HISTORY_DAYS))
            c_end = None if random.random() < 0.15 else c_start + timedelta(days=random.randint(180, 1460))
            c_created = ts_on(c_start - timedelta(days=random.randint(0, 30)))
            contract_over = c_end is not None and c_end < REFERENCE_DATE
            # a finished project ends inside its contract and before the reference day;
            # projects start early enough to leave room for that where possible
            last_end = REFERENCE_DATE - timedelta(days=1)
            if c_end is not None:
                last_end = min(last_end, c_end)
            latest_start = max(c_start, last_end - timedelta(days=30))
            names: set[str] = set()
            budget_sum = 0

            for _ in range(counts[contract_id - 1]):
                project_id += 1
                p_start = day_between(c_start, latest_start)
                room = (last_end - p_start).days
                outcome = random.choices(OUTCOMES, OUTCOME_WEIGHTS)[0]
                if contract_over and outcome not in FINISHED:
                    # nothing is still running under an expired contract
                    outcome = random.choices(OUTCOMES[:2], OUTCOME_WEIGHTS[:2])[0]
                elif outcome in FINISHED and room < PROJECT_DAYS[outcome][0]:
                    # the contract is too young for the project to have finished
                    outcome = "In Progress"
                path = status_path(outcome)
                first = ts_on(p_start)
                if outcome in FINISHED:
                    min_days, max_days = PROJECT_DAYS[outcome]
                    p_end = p_start + timedelta(days=random.randint(min_days, min(max_days, room)))
                    window_end = ts_on(p_end)
                    stamps = [first, *ts_strictly_between(first, window_end, len(path) - 2), window_end]
                else:
                    p_end = None
                    window_end = REFERENCE_END
                    stamps = [first, *ts_strictly_between(first, window_end, len(path) - 1)]

                # budget chain: each audit row starts where the previous one ended,
                # and the last new_budget is the project's current budget
                budgets = [draw_budget()]
                for _ in range(random.choices([0, 1, 2, 3], weights=[50, 30, 15, 5])[0]):
                    budgets.append(revised_budget(budgets[-1]))
                changes = ts_strictly_between(first, window_end, len(budgets) - 1)
                for old_budget, new_budget, changed in zip(budgets, budgets[1:], changes):
                    audit_id += 1
                    audits.writerow([audit_id, project_id, old_budget, new_budget, changed, changed])
                budget_sum += budgets[-1]

                name = fake.catch_phrase()
                while name in names:
                    name = fake.catch_phrase()
                names.add(name)
                updated = max(stamps[-1], changes[-1]) if changes else stamps[-1]
                projects.writerow([project_id, contract_id, STATUS_ID[path[-1]], name, p_start, p_end,
                                   budgets[-1], first, updated])

                for tech_id in random.sample(TECH_IDS, random.randint(1, 5)):
                    technologies.writerow([project_id, tech_id])
                    n_technologies += 1

                for status, changed in zip(path, stamps):
                    history_id += 1
                    if random.random() < 0.85:
                        actor = (random.choice(vendor_users), None)
                    else:
                        actor = (None, random.choice(mgmt_user_ids))
                    comment = random.choice(STATUS_COMMENTS[status]) if random.random() < 0.65 else None
                    history.writerow([history_id, project_id, *actor, STATUS_ID[status], changed, comment])

                if outcome == "Completed":
                    completed_pid.append(project_id)
                    completed_cid.append(contract_id)
                    completed_end.append(p_end.toordinal())
                if project_id % PROGRESS_EVERY == 0:
                    print(".", end="", flush=True)

            # seed convention: a stated framework value exceeds the sum of the current
            # project budgets; some contracts state none
            total_value = None if random.random() < 0.10 else money(budget_sum * random.uniform(1.05, 1.5))
            contracts.writerow([
                contract_id, client_id, vendor_id, f"CTR-{contract_id:08d}", fake.bs().title(), c_start, c_end,
                total_value, random.choices(CURRENCIES, CURRENCY_WEIGHTS)[0], fake.paragraph(nb_sentences=2),
                c_end is None or c_end >= REFERENCE_DATE, c_created, c_created,
            ])
            contract_client.append(client_id)
            contract_vendor.append(vendor_id)
    print()
    report("client_vendor_contract.csv", n_contracts, t0)
    report("project.csv", project_id, t0)
    report("project_technology.csv", n_technologies, t0)
    report("project_budget_audit.csv", audit_id, t0)
    report("project_status_history.csv", history_id, t0)
    return contract_client, contract_vendor, (completed_pid, completed_cid, completed_end)


# ═════════════════════════════════════════════════════════════════════════════
# 6. REVIEWS, SCORES & DISPUTE TICKETS
# ═════════════════════════════════════════════════════════════════════════════
REVIEW_COLUMNS = [
    "review_id", "project_id", "client_user_id", "review_date", "summary_text", "is_published",
    "created_at", "updated_at",
]
DISPUTE_COLUMNS = [
    "ticket_id", "assigned_management_user_id", "review_id", "vendor_user_id", "reason", "is_resolved",
    "filed_at", "resolved_at", "resolution_note", "created_at", "updated_at",
]


def write_reviews(
    out_dir: Path,
    n_reviews: int,
    completed: tuple[array, array, array],
    contract_client: array,
    contract_vendor: array,
    client_to_users: list[list[int]],
    vendor_to_users: list[list[int]],
    mgmt_user_ids: list[int],
    vendor_quality: list[float],
) -> None:
    """Reviews go to a sample of the completed projects, 1-45 days after the end
    date, written by a user of the project's client. A vendor user of the project's
    vendor may dispute the review; while that ticket is open the review stays
    unpublished, otherwise about 70% of the reviews are published."""
    t0 = time.time()
    completed_pid, completed_cid, completed_end = completed
    if n_reviews > len(completed_pid):
        print(f"     only {len(completed_pid):,} completed projects: writing that many reviews")
        n_reviews = len(completed_pid)
    chosen = sorted(random.sample(range(len(completed_pid)), n_reviews))
    ticket_id = 0
    print(f"     dot = {PROGRESS_EVERY:,} reviews:  ", end="", flush=True)
    with (
        csv_file(out_dir, "review.csv", REVIEW_COLUMNS) as reviews,
        csv_file(out_dir, "review_score.csv", ["review_id", "dimension_id", "score_value"]) as scores,
        csv_file(out_dir, "dispute_ticket.csv", DISPUTE_COLUMNS) as disputes,
    ):
        for review_id, index in enumerate(chosen, start=1):
            project_id, contract_id = completed_pid[index], completed_cid[index]
            client_id, vendor_id = contract_client[contract_id - 1], contract_vendor[contract_id - 1]
            end_date = date.fromordinal(completed_end[index])
            review_date = end_date + timedelta(days=random.randint(1, min(45, (REFERENCE_DATE - end_date).days)))
            created = ts_on(review_date)

            # the dispute is decided first, because it decides whether the review is published;
            # seed disputes are filed at least one day after the review
            days_left = (REFERENCE_DATE - review_date).days
            disputed = days_left >= 1 and random.random() < DISPUTE_RATE
            resolved_at = None
            if disputed:
                filed_at = review_date + timedelta(days=random.randint(1, min(30, days_left)))
                resolved_day = filed_at + timedelta(days=random.randint(1, 120))
                if resolved_day <= REFERENCE_DATE and random.random() < 0.85:
                    resolved_at = ts_on(resolved_day)
            is_published = (not disputed or resolved_at is not None) and random.random() < 0.70
            updated = ts_after(max(created, resolved_at or created), 14) if is_published else created
            author_id = random.choice(client_to_users[client_id - 1])
            reviews.writerow([review_id, project_id, author_id, review_date,
                              fake.paragraph(nb_sentences=3), is_published, created, updated])

            quality = vendor_quality[vendor_id - 1]
            for dimension_id, _, _, bias in DIMENSIONS:
                score = round(min(5.0, max(1.0, random.gauss(quality + bias, 0.8))))
                scores.writerow([review_id, dimension_id, score])

            if disputed:
                ticket_id += 1
                resolved = resolved_at is not None
                filed = ts_on(filed_at)
                disputes.writerow([
                    ticket_id, random.choice(mgmt_user_ids) if resolved else None, review_id,
                    random.choice(vendor_to_users[vendor_id - 1]), fake.paragraph(nb_sentences=2), resolved,
                    filed_at, resolved_at, fake.sentence() if resolved else None, filed, resolved_at or filed,
                ])
            if review_id % PROGRESS_EVERY == 0:
                print(".", end="", flush=True)
    print()
    report("review.csv", n_reviews, t0)
    report("review_score.csv", n_reviews * len(DIMENSIONS), t0)
    report("dispute_ticket.csv", ticket_id, t0)


# ═════════════════════════════════════════════════════════════════════════════
# MAIN
# ═════════════════════════════════════════════════════════════════════════════
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate the VQEMS seed data as CSV files for load_seed.sql.")
    parser.add_argument("out_dir", nargs="?", default="seed_data", type=Path,
                        help="directory for the CSV files (default: seed_data)")
    parser.add_argument("--scale", type=positive_float, default=1.0,
                        help="multiplies every table size; lookup tables are fixed (default: 1.0)")
    parser.add_argument("--seed", type=int, default=42, help="random seed (default: 42)")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    Faker.seed(args.seed)
    random.seed(args.seed)
    out_dir: Path = args.out_dir
    out_dir.mkdir(parents=True, exist_ok=True)

    n_vendors = scaled(NUM_VENDORS, args.scale)
    n_clients = scaled(NUM_CLIENTS, args.scale)
    n_client_users = scaled(NUM_CLIENT_USERS, args.scale)
    n_vendor_users = scaled(NUM_VENDOR_USERS, args.scale)
    n_mgmt_users = scaled(NUM_MGMT_USERS, args.scale)
    n_contracts = scaled(NUM_CONTRACTS, args.scale)
    n_projects = scaled(NUM_PROJECTS, args.scale)
    n_reviews = scaled(NUM_REVIEWS, args.scale)
    print(f"Seed {args.seed}, scale {args.scale:g}, reference date {REFERENCE_DATE}: "
          f"{n_vendors:,} vendors, {n_clients:,} clients, {n_client_users + n_vendor_users + n_mgmt_users:,} users, "
          f"{n_contracts:,} contracts, {n_projects:,} projects, {n_reviews:,} reviews")

    print("\n── 1. Lookup tables ──────────────────────────────────────────────────")
    write_lookup_tables(out_dir)

    print("\n── 2. Vendors & Clients ──────────────────────────────────────────────")
    vendor_domains, vendor_quality = write_vendors(out_dir, n_vendors)
    client_domains = write_clients(out_dir, n_clients)

    print("\n── 3. Users ──────────────────────────────────────────────────────────")
    client_to_users, vendor_to_users, mgmt_user_ids = write_users(
        out_dir, n_client_users, n_vendor_users, n_mgmt_users, client_domains, vendor_domains
    )

    print("\n── 4. Vendor subscriptions ───────────────────────────────────────────")
    write_subscriptions(out_dir, n_vendors)

    print("\n── 5. Contracts & Projects ───────────────────────────────────────────")
    contract_client, contract_vendor, completed = write_contracts_and_projects(
        out_dir, n_contracts, n_projects, n_clients, n_vendors, vendor_to_users, mgmt_user_ids
    )

    print("\n── 6. Reviews, scores & disputes ─────────────────────────────────────")
    write_reviews(
        out_dir, n_reviews, completed, contract_client, contract_vendor,
        client_to_users, vendor_to_users, mgmt_user_ids, vendor_quality,
    )

    print(f"\n  All CSV files written to: {out_dir.resolve()}")


if __name__ == "__main__":
    main()