#!/usr/bin/env python3 """为客户+品类证件额度联调造数（测试库）。 造：客户、报价（accepted）、合同（executing）+ 合同产品、 2 张购买证（10吨早到期 / 100吨晚到期）、2 张运输证（引用对应购买证）、期初 opening。 不造：品类↔系统产品关联（由用户在界面配置）。 用法： cd backend && source venv/bin/activate PYTHONPATH=. python scripts/seed_pl_customer_cert_test_data.py --dry-run PYTHONPATH=. python scripts/seed_pl_customer_cert_test_data.py --apply """ from __future__ import annotations import argparse import json import os import sys from datetime import date, datetime, timedelta from decimal import Decimal from pathlib import Path from urllib.parse import unquote, urlparse from uuid import uuid4 BACKEND_ROOT = Path(__file__).resolve().parents[1] TENANT = 'shanxia2.0' # 固定前缀，便于识别/清理 TAG = 'PL-CERT-SEED-20261114' CUSTOMER_NAME = 'PL证件额度测试客户' PRODUCT_ID = '339E64E02E7C41F3A04AACCE71E699BE' # 98% 工业硫酸 is_toxic=1 on OWNER_USER_ID = '02584218205246EEB5F2AC96DF465A9D' OWNER_NAME = '方金强' SELLER = '安庆市山峡化工有限公司' VEHICLE = '皖A9TEST1' def load_url(): env_file = os.environ.get('ENV_FILE') or str(BACKEND_ROOT / '.env') for line in Path(env_file).read_text(encoding='utf-8').splitlines(): if line.startswith('DATABASE_URL='): return line.split('=', 1)[1].strip().strip('"').strip("'") raise SystemExit('DATABASE_URL missing') def connect(): import pymysql raw = load_url().replace('mysql+pymysql://', 'mysql://', 1) p = urlparse(raw) return pymysql.connect( host=p.hostname, port=p.port or 3306, user=unquote(p.username or ''), password=unquote(p.password or ''), database=(p.path or '/').lstrip('/'), charset='utf8mb4', autocommit=False, ) def uid(suffix: str) -> str: # 稳定 ID，重复执行可先删再插 return f'{TAG}-{suffix}' def main() -> int: parser = argparse.ArgumentParser() g = parser.add_mutually_exclusive_group(required=True) g.add_argument('--dry-run', action='store_true') g.add_argument('--apply', action='store_true') args = parser.parse_args() apply = args.apply now = datetime.now().replace(microsecond=0) today = date.today() ids = { 'customer': uid('customer'), 'quote': uid('quote'), 'quote_product': uid('quote-prod'), 'contract': uid('contract'), 'contract_product': uid('contract-prod'), 'purchase_early': uid('purchase-10t'), 'purchase_late': uid('purchase-100t'), 'transport_early': uid('transport-10t'), 'transport_late': uid('transport-100t'), 'ref_early': uid('ref-10t'), 'ref_late': uid('ref-100t'), 'open_pe': uid('open-p-10t'), 'open_pl': uid('open-p-100t'), 'open_te': uid('open-t-10t'), 'open_tl': uid('open-t-100t'), } conn = connect() cur = conn.cursor() try: # category for 98% 工业硫酸 cur.execute( "SELECT id FROM pl_cert_category WHERE tenant_no=%s AND del_flag=0 AND category_name=%s", (TENANT, '98% 工业硫酸'), ) row = cur.fetchone() if not row: raise SystemExit('缺少品类「98% 工业硫酸」，请先跑 backfill 或手工创建') category_id = row[0] cur.execute( "SELECT product_no, product_name, unit, COALESCE(price,1000) FROM crm_product WHERE id=%s AND del_flag=0", (PRODUCT_ID,), ) prod = cur.fetchone() if not prod: raise SystemExit(f'产品不存在: {PRODUCT_ID}') product_no, product_name, unit, price = prod price = Decimal(price) qty = Decimal('100') subtotal = (price * qty).quantize(Decimal('0.01')) print('=== seed plan ===') print(f'customer={CUSTOMER_NAME} id={ids["customer"]}') print(f'product={product_name} ({PRODUCT_ID}) toxic unit={unit}') print(f'category_id={category_id}') print(f'vehicle={VEHICLE}') print(f'purchase early 10 / late 100 ; transport pair each') # cleanup previous seed with same ids cleanup_tables = [ ('pl_cert_qty_log', 'id', [ids['open_pe'], ids['open_pl'], ids['open_te'], ids['open_tl']]), ('pl_transport_cert_purchase_ref', 'id', [ids['ref_early'], ids['ref_late']]), ('pl_transport_cert', 'id', [ids['transport_early'], ids['transport_late']]), ('pl_purchase_cert', 'id', [ids['purchase_early'], ids['purchase_late']]), ('crm_contract_product', 'id', [ids['contract_product']]), ('crm_contract', 'id', [ids['contract']]), ('crm_quote_product', 'id', [ids['quote_product']]), ('crm_quote', 'id', [ids['quote']]), ('crm_customer', 'id', [ids['customer']]), ] for table, col, values in cleanup_tables: for vid in values: cur.execute(f"DELETE FROM {table} WHERE {col}=%s", (vid,)) # customer cur.execute( """ INSERT INTO crm_customer (id, customer_no, customer_name, status, deal_status, owner_user_id, owner_user_name, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,'normal','undeal',%s,%s,%s,'',%s,%s,%s,0) """, (ids['customer'], f'C-{TAG}', CUSTOMER_NAME, OWNER_USER_ID, OWNER_NAME, TENANT, OWNER_USER_ID, now, now), ) print('+ customer') # quote quote_no = f'BJ-{TAG}' cur.execute( """ INSERT INTO crm_quote (id, quote_no, quote_title, quote_type, customer_id, customer_name, owner_user_id, version, is_latest, total_amount, final_amount, quote_date, expire_date, status, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,'sales',%s,%s,%s,1,1,%s,%s,%s,%s,'accepted',%s,'',%s,%s,%s,0) """, (ids['quote'], quote_no, f'{CUSTOMER_NAME}-硫酸报价', ids['customer'], CUSTOMER_NAME, OWNER_USER_ID, subtotal, subtotal, today, today + timedelta(days=30), TENANT, OWNER_USER_ID, now, now), ) cur.execute( """ INSERT INTO crm_quote_product (id, quote_id, product_id, product_name, product_no, unit, quantity, price, discount, subtotal, sort, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,100,%s,1,%s,'',%s,%s,%s,0) """, (ids['quote_product'], ids['quote'], PRODUCT_ID, product_name, product_no, unit, qty, price, subtotal, TENANT, OWNER_USER_ID, now, now), ) print('+ quote', quote_no) # contract contract_no = f'HT-{TAG}' cur.execute( """ INSERT INTO crm_contract (id, contract_no, contract_name, customer_id, customer_name, quote_id, contract_type, amount, currency, tax_rate, status, supplier_id, party_a, party_b, start_date, end_date, sign_date, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,%s,%s,%s,'sale',%s,'CNY',13,'executing','1',%s,%s,%s,%s,%s, %s,'',%s,%s,%s,0) """, (ids['contract'], contract_no, f'{CUSTOMER_NAME}-购销合同', ids['customer'], CUSTOMER_NAME, ids['quote'], subtotal, SELLER, CUSTOMER_NAME, today, today + timedelta(days=365), today, TENANT, OWNER_USER_ID, now, now), ) cur.execute( """ INSERT INTO crm_contract_product (id, contract_id, product_id, product_name, product_no, unit, quantity, price, tax_rate, subtotal, delivery_status, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,13,%s,'pending',%s,'',%s,%s,%s,0) """, (ids['contract_product'], ids['contract'], PRODUCT_ID, product_name, product_no, unit, qty, price, subtotal, TENANT, OWNER_USER_ID, now, now), ) print('+ contract', contract_no, 'product_row', ids['contract_product']) def insert_purchase(pid, cert_no, total, valid_to, face='98% 工业硫酸'): cur.execute( """ INSERT INTO pl_purchase_cert (id, cert_no, issue_date, product_name_on_cert, total_qty, quantity_unit, valid_from, valid_to, valid_use_text, entry_mode, status, verified_by, verified_time, reserved_qty, used_qty, opening_single_use_consumed, revision, buyer_name, seller_name, customer_id, category_id, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,%s,%s,'dun',%s,%s,'多次有效','manual','active', %s,%s,0,0,0,1,%s,%s,%s,%s,%s,'',%s,%s,%s,0) """, (pid, cert_no, today, face, Decimal(total), today, valid_to, OWNER_USER_ID, now, CUSTOMER_NAME, SELLER, ids['customer'], category_id, TENANT, OWNER_USER_ID, now, now), ) op = str(uuid4()) cur.execute( """ INSERT INTO pl_cert_qty_log (id, operation_id, request_hash, cert_type, cert_id, event_type, quantity_unit, total_delta, reserved_delta, used_delta, before_total_qty, after_total_qty, before_reserved_qty, after_reserved_qty, before_used_qty, after_used_qty, reason, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,'purchase',%s,'opening','dun', 0,0,0,%s,%s,0,0,0,0,'seed opening',%s,'',%s,%s,%s,0) """, (str(uuid4()) if pid.endswith('10t') else str(uuid4()), op, op, pid, Decimal(total), Decimal(total), TENANT, OWNER_USER_ID, now, now), ) # fix opening ids to stable ones def insert_opening(log_id, cert_type, cert_id, total): op = log_id cur.execute( """ INSERT INTO pl_cert_qty_log (id, operation_id, request_hash, cert_type, cert_id, event_type, quantity_unit, total_delta, reserved_delta, used_delta, before_total_qty, after_total_qty, before_reserved_qty, after_reserved_qty, before_used_qty, after_used_qty, reason, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,%s,%s,%s,'opening','dun', 0,0,0,%s,%s,0,0,0,0,'seed opening',%s,'',%s,%s,%s,0) """, (log_id, op, op, cert_type, cert_id, Decimal(total), Decimal(total), TENANT, OWNER_USER_ID, now, now), ) # rewrite purchase insert without nested opening def insert_purchase2(pid, cert_no, total, valid_to, open_id): cur.execute( """ INSERT INTO pl_purchase_cert (id, cert_no, issue_date, product_name_on_cert, total_qty, quantity_unit, valid_from, valid_to, valid_use_text, entry_mode, status, verified_by, verified_time, reserved_qty, used_qty, opening_single_use_consumed, revision, buyer_name, seller_name, customer_id, category_id, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag, remark) VALUES (%s,%s,%s,'98%% 工业硫酸',%s,'dun',%s,%s,'多次有效','manual','active', %s,%s,0,0,0,1,%s,%s,%s,%s,%s,'',%s,%s,%s,0,%s) """, (pid, cert_no, today, Decimal(total), today, valid_to, OWNER_USER_ID, now, CUSTOMER_NAME, SELLER, ids['customer'], category_id, TENANT, OWNER_USER_ID, now, now, TAG), ) insert_opening(open_id, 'purchase', pid, total) def insert_transport2(tid, cert_no, total, valid_to, open_id, purchase_id, ref_id): plates = json.dumps([VEHICLE], ensure_ascii=False) cur.execute( """ INSERT INTO pl_transport_cert (id, cert_no, issue_date, product_name_on_cert, total_qty, quantity_unit, valid_from, valid_to, valid_use_text, entry_mode, status, verified_by, verified_time, reserved_qty, used_qty, opening_single_use_consumed, revision, shipper_name, consignee_name, vehicle_numbers_json, customer_id, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag, remark) VALUES (%s,%s,%s,'98%% 工业硫酸',%s,'dun',%s,%s,'多次有效','manual','active', %s,%s,0,0,0,1,%s,%s,%s,%s,%s,'',%s,%s,%s,0,%s) """, (tid, cert_no, today, Decimal(total), today, valid_to, OWNER_USER_ID, now, SELLER, CUSTOMER_NAME, plates, ids['customer'], TENANT, OWNER_USER_ID, now, now, TAG), ) insert_opening(open_id, 'transport', tid, total) cur.execute( """ INSERT INTO pl_transport_cert_purchase_ref (id, transport_cert_id, reference_type, reference_no, purchase_cert_id, tenant_no, create_dept_id, create_by, create_time, update_time, del_flag) VALUES (%s,%s,'purchase_permit',%s,%s,%s,'',%s,%s,%s,0) """, (ref_id, tid, cert_no.replace('Y-', 'G-'), purchase_id, TENANT, OWNER_USER_ID, now, now), ) early_to = today + timedelta(days=30) late_to = today + timedelta(days=180) insert_purchase2(ids['purchase_early'], f'G-{TAG}-10T', 10, early_to, ids['open_pe']) insert_purchase2(ids['purchase_late'], f'G-{TAG}-100T', 100, late_to, ids['open_pl']) insert_transport2( ids['transport_early'], f'Y-{TAG}-10T', 10, early_to, ids['open_te'], ids['purchase_early'], ids['ref_early'], ) insert_transport2( ids['transport_late'], f'Y-{TAG}-100T', 100, late_to, ids['open_tl'], ids['purchase_late'], ids['ref_late'], ) print('+ purchase x2 + transport x2 + opening + refs') print('--- ids ---') for k, v in ids.items(): print(f'{k}: {v}') print('--- how to test ---') print(f'1. 证用品类「98% 工业硫酸」关联产品 {PRODUCT_ID}（{product_name}）') print(f'2. 新建运输计划：客户={CUSTOMER_NAME}，产品用上述合同行') print(f'3. 安排车辆请用车号 {VEHICLE}（或把运输证车号改成你的车）') print('4. 测规则A：计划量 50 吨时，应跳过 10 吨早到期证，选用 100 吨证') if apply: conn.commit() print('COMMITTED') else: conn.rollback() print('DRY-RUN rolled back') return 0 except Exception: conn.rollback() raise finally: cur.close() conn.close() if __name__ == '__main__': raise SystemExit(main())