utils/gen_test_data.py - master - unTonee/LekalaOrderManager

utils/gen_test_data.py
239 строк10 KB
Anton Kostrubin aka unTonee
chore: примеры файлов и утилиты
30 июн 2026, 00:36
30 июн 2026, 00:36208eabe
import os
import random
import uuid
import math
import sqlite3
from datetime import datetime, timedelta
from collections import defaultdict

try:
    from tqdm import tqdm
except ImportError:
    print("⚠️  tqdm не установлен. Установите: pip install tqdm")
    # Fallback simple progress
    class tqdm:
        def __init__(self, iterable=None, total=None, desc=None):
            self.iterable = iterable
            self.total = total or len(iterable) if iterable else 0
            self.desc = desc
            self.n = 0
        def __iter__(self):
            for item in self.iterable:
                self.n += 1
                if self.n % 50 == 0: print(f"  {self.desc}: {self.n}/{self.total}")
                yield item
        def update(self, n=1): self.n += n

# === КОНФИГУРАЦИЯ ===
SEED = 42
START_DATE = datetime(2026, 6, 20, 8, 0, 0)
END_DATE = datetime(2026, 7, 20, 20, 0, 0)
WORK_HOURS = (8, 20)
OUTPUT_SQL = "test_data_dump.sql"
DB_PATH = "data/app.db"
UPLOAD_DIR = "app/uploads"

USERS = {
    4: {"name": "Иванова Е.А.", "role": "manager", "weight": 0.18},
    5: {"name": "Смирнов Д.В.", "role": "manager", "weight": 0.18},
    6: {"name": "Попова А.М.", "role": "manager", "weight": 0.09},
    7: {"name": "Соколов К.А.", "role": "manager", "weight": 0.09},
    8: {"name": "Морозова Е.В.", "role": "manager", "weight": 0.09},
    9: {"name": "Волков И.С.", "role": "manager", "weight": 0.09},
    10: {"name": "Кузнецова О.Л.", "role": "manager", "weight": 0.09},
    11: {"name": "Белов П.А.", "role": "manager", "weight": 0.09},
    12: {"name": "Васильева Н.С.", "role": "manager", "weight": 0.09},
    13: {"name": "Зайцев А.Ю.", "role": "supervisor", "weight": 0.0},
}
MANAGER_IDS = [uid for uid, u in USERS.items() if u["role"] == "manager"]
SUPERVISOR_ID = 13

CLIENT_IDS = list(range(1, 25))
MATERIAL_IDS = list(range(100, 120))
DELIVERY_IDS = list(range(1, 7))
DISCOUNT_IDS = [None, None, None, 1, 2]
PAY_TYPE_IDS = [1, 2, 3]

stats = defaultdict(int)
daily_orders = defaultdict(int)
user_orders = defaultdict(int)
status_counts = defaultdict(int)

random.seed(SEED)

def get_work_datetime(start, end):
    while True:
        delta = (end - start).total_seconds()
        dt = start + timedelta(seconds=random.uniform(0, delta))
        if dt.weekday() != 6 and WORK_HOURS[0] <= dt.hour < WORK_HOURS[1]:
            return dt

def create_fake_file():
    ext = random.choice(["pdf", "plt", "plo"])
    server_name = f"{uuid.uuid4().hex}.{ext}"
    path = os.path.join(UPLOAD_DIR, server_name)
    os.makedirs(UPLOAD_DIR, exist_ok=True)
    with open(path, "wb") as f:
        f.write(b"%PDF-1.4 fake" if ext == "pdf" else b"FAKE PATTERN DATA")
    stats["files_created"] += 1
    return server_name

def validate_database(db_path):
    """Проверка целостности данных после импорта"""
    print("\n🔍 ЗАПУСК ВАЛИДАЦИИ ДАННЫХ...")
    checks = [
        ("Заказы с отрицательным итогом", 
         "SELECT COUNT(*) FROM orders WHERE total_sum < 0"),
        ("Строки без снапшота материала", 
         "SELECT COUNT(*) FROM order_items WHERE material_name_snapshot IS NULL"),
        ("Заказы без автора", 
         "SELECT COUNT(*) FROM orders WHERE author IS NULL"),
        ("Файлы лекал в БД существуют на диске", 
         None),  # Специальная проверка
        ("Платежи > суммы заказа (переплата корректна)", 
         "SELECT COUNT(*) FROM payments p JOIN orders o ON p.order_id=o.id WHERE p.amount > o.total_sum * 5"),
        ("Системные записи не удалены", 
         "SELECT COUNT(*) FROM nomenclature WHERE id IN (90001,90002) AND to_recycle=1"),
    ]
    
    conn = sqlite3.connect(db_path)
    cur = conn.cursor()
    all_passed = True
    
    for name, query in checks:
        if query:
            count = cur.execute(query).fetchone()[0]
            status = "✅ PASS" if count == 0 else f"❌ FAIL ({count})"
            if count > 0: all_passed = False
        else:
            # Проверка файлов
            missing = cur.execute(
                "SELECT pattern_server_name FROM order_items WHERE pattern_server_name IS NOT NULL"
            ).fetchall()
            bad_files = sum(1 for (fn,) in missing if not os.path.exists(os.path.join(UPLOAD_DIR, fn)))
            status = "✅ PASS" if bad_files == 0 else f"❌ FAIL ({bad_files} missing)"
            if bad_files > 0: all_passed = False
            
        print(f"  {name:<45} {status}")
    
    conn.close()
    return all_passed

def generate_sql():
    sql_lines = [
        "-- AUTO GENERATED TEST DATA",
        f"-- Period: {START_DATE.date()} to {END_DATE.date()}",
        "PRAGMA foreign_keys=OFF;",
        "BEGIN TRANSACTION;"
    ]
    
    total_days = (END_DATE - START_DATE).days
    target_orders_per_day = random.randint(60, 90)
    total_orders = target_orders_per_day * total_days
    
    print(f"🏭 Генерация ~{total_orders} заказов...")
    orders_data = []
    
    for i in tqdm(range(total_orders), desc="📦 Планирование заказов"):
        manager_id = random.choices(MANAGER_IDS, weights=[USERS[m]["weight"] for m in MANAGER_IDS])[0]
        created_at = get_work_datetime(START_DATE, END_DATE)
        daily_orders[created_at.date()] += 1
        user_orders[manager_id] += 1
        
        client_id = random.choice(CLIENT_IDS)
        delivery_id = random.choice(DELIVERY_IDS)
        discount_id = random.choice(DISCOUNT_IDS)
        is_asap = random.random() < 0.12
        is_problem = random.random() < 0.025
        
        items_count = random.randint(1, 5)
        items_sql = []
        for line_num in range(1, items_count + 1):
            server_name = create_fake_file()
            w, h = round(random.uniform(0.5, 2.5), 4), round(random.uniform(0.5, 2.5), 4)
            mat_id = random.choice(MATERIAL_IDS)
            qty = round(random.uniform(1, 10), 2)
            has_adj = random.random() < 0.3
            has_glue = random.random() < 0.2
            
            item_sql = f"""
INSERT INTO v_order_items (order_id, line_number, pattern_name, pattern_width, pattern_height, 
                           pattern_server_name, material_id, quantity, has_adjustment, adjustment_qty,
                           has_gluing, gluing_qty)
VALUES ({{ORDER_ID}}, {line_num}, 'pattern_{line_num}.plt', {w}, {h}, 
        '{server_name}', {mat_id}, {qty}, {int(has_adj)}, {round(w*h*0.1, 2) if has_adj else 0},
        {int(has_glue)}, {math.ceil(w*h)});"""
            items_sql.append(item_sql)
        
        final_status = 6 if is_problem else random.choice([3,4,5] if (datetime.now()-created_at).days > 3 else [1,2,3])
        to_recycle = 1 if (is_problem and random.random() < 0.5) else 0
        status_counts[final_status] += 1
        
        disc_clause = f"{discount_id}" if discount_id else "NULL"
        order_sql = f"""
INSERT INTO v_orders (created_at, client_id, delivery_list_id, delivery_price, status_id, 
                      discount_id, author, note, asap, to_recycle)
VALUES ('{created_at.strftime('%Y-%m-%d %H:%M:%S')}', {client_id}, {delivery_id}, 
        {round(random.uniform(0, 800), 2)}, {final_status}, 
        {disc_clause}, {manager_id}, 'Тестовый заказ #{i+1}', {int(is_asap)}, {to_recycle});"""
        
        orders_data.append({
            "sql": order_sql, "items": items_sql, "created_at": created_at,
            "manager_id": manager_id, "is_problem": is_problem
        })
    
    orders_data.sort(key=lambda x: x["created_at"])
    
    print("💾 Формирование SQL дампа...")
    for idx, od in enumerate(tqdm(orders_data, desc="✍️  Запись SQL")):
        oid_func = "(SELECT COALESCE(MAX(id),0)+1 FROM orders)" if idx == 0 else "(SELECT MAX(id) FROM orders)"
        sql_lines.append(od["sql"].replace("{ORDER_ID}", oid_func))
        for item_sql in od["items"]:
            sql_lines.append(item_sql.replace("{ORDER_ID}", "(SELECT MAX(id) FROM orders)"))
            
        if not od["is_problem"] and random.random() < 0.8:
            pay_date = od["created_at"] + timedelta(hours=random.uniform(1, 48))
            sql_lines.append(f"""
INSERT INTO payments (order_id, pay_type_id, amount, payment_date, comment)
VALUES ((SELECT MAX(id) FROM orders), {random.choice(PAY_TYPE_IDS)}, 
        ROUND(RANDOM()*5000+500, 2), '{pay_date.strftime('%Y-%m-%d %H:%M:%S')}', 'Оплата менеджером');""")
            stats["payments"] += 1
            
        if random.random() < 0.4:
            comm_user = SUPERVISOR_ID if random.random() < 0.3 else od["manager_id"]
            sql_lines.append(f"""
INSERT INTO order_comments (order_id, user_id, text, created_at)
VALUES ((SELECT MAX(id) FROM orders), {comm_user}, 'Уточнение по файлам', '{od["created_at"].strftime('%Y-%m-%d %H:%M:%S')}');""")
            stats["comments"] += 1
    
    sql_lines.extend(["COMMIT;", "PRAGMA foreign_keys=ON;"])
    
    with open(OUTPUT_SQL, "w", encoding="utf-8") as f:
        f.write("\n".join(sql_lines))
    
    print(f"\n✅ SQL сохранен: {OUTPUT_SQL}")
    print(f"📁 Файлы: {os.path.abspath(UPLOAD_DIR)}")
    return len(orders_data)

if __name__ == "__main__":
    total = generate_sql()
    
    # Автоимпорт и валидация если есть БД
    if os.path.exists(DB_PATH):
        print("\n🔄 Импорт в базу данных...")
        os.system(f'sqlite3 "{DB_PATH}" < "{OUTPUT_SQL}"')
        validate_database(DB_PATH)
    
    # Сводка
    print("\n" + "="*60)
    print("📊 СВОДКА ПО ТЕСТОВЫМ ДАННЫМ")
    print("="*60)
    print(f"{'Всего заказов':<35} {total:>10}")
    print(f"{'Файлов лекал':<35} {stats['files_created']:>10}")
    print(f"{'Платежей':<35} {stats['payments']:>10}")
    print(f"{'Комментариев':<35} {stats['comments']:>10}")
    print("-"*60)
    print("\n👥 НАГРУЗКА МЕНЕДЖЕРОВ:")
    for uid in sorted(user_orders.keys()):
        bar = "█" * (user_orders[uid] // 10)
        print(f"  {USERS[uid]['name']:<20} {user_orders[uid]:>4} {bar}")