"""店铺批量导入 SQL 构造。""" from __future__ import annotations from datetime import datetime from hashlib import sha256 from .shop_import import ShopImportPlan, ShopImportRow, initial_password_hash def build_shop_sql(plan: ShopImportPlan) -> str: """生成包含店铺初始事实的单事务 SQL。""" lines = [ "-- 店铺批量导入 SQL；执行前必须审核结果清单与摘要。", f"-- 批次: {plan.config.batch_id}", f"-- 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "-- 建议执行: psql -1 -v ON_ERROR_STOP=1 -f <此文件>", "", "-- psql -1 开启的事务内先执行以下守卫。", "", *_preflight_sql(plan), ] for row in plan.rows: lines.extend(_row_sql(row, plan)) lines.append("") return "\n".join(lines) def _preflight_sql(plan: ShopImportPlan) -> list[str]: config = plan.config mapped_salespeople = [] seen_salespeople: set[str] = set() for row in plan.rows: if row.salesperson not in seen_salespeople: seen_salespeople.add(row.salesperson) mapped_salespeople.append((row.line_no, row.salesperson, config.salesperson_account_ids[row.salesperson])) target_rows = ",\n ".join( f"({row.line_no}, {_sql_str(row.shop_code)}, {_sql_str(row.username)}, {_sql_str(row.phone)})" for row in plan.rows ) salesperson_rows = ",\n ".join( f"({line_no}, {_sql_str(name)}, {account_id})" for line_no, name, account_id in mapped_salespeople ) return [ "-- 所有目标库守卫必须在业务写入前完成。", "DO $shop_import_guard$", "DECLARE", " v_row record;", "BEGIN", " IF NOT EXISTS (SELECT 1 FROM tb_role WHERE id = " + str(config.default_role_id) + " AND role_type = 2 AND status = 1 AND deleted_at IS NULL) THEN", f" RAISE EXCEPTION '导入配置错误: default_role_id={config.default_role_id} 不是启用的客户角色';", " END IF;", " IF NOT EXISTS (SELECT 1 FROM tb_account WHERE id = " + str(config.operator_id) + " AND user_type = 1 AND status = 1 AND deleted_at IS NULL) THEN", f" RAISE EXCEPTION '导入配置错误: operator_id={config.operator_id} 不是启用的超级管理员';", " END IF;", " FOR v_row IN", " VALUES", f" {salesperson_rows}", " LOOP", " IF NOT EXISTS (SELECT 1 FROM tb_account WHERE id = v_row.column3 AND user_type = 2 AND status = 1 AND deleted_at IS NULL) THEN", " RAISE EXCEPTION 'CSV 第 % 行业务员映射无效: % (账号 ID=%)', v_row.column1, v_row.column2, v_row.column3;", " END IF;", " END LOOP;", " FOR v_row IN", " VALUES", f" {target_rows}", " LOOP", " IF EXISTS (SELECT 1 FROM tb_shop WHERE shop_code = v_row.column2 AND deleted_at IS NULL) THEN", " RAISE EXCEPTION 'CSV 第 % 行店铺编号已存在: %', v_row.column1, v_row.column2;", " END IF;", " IF EXISTS (SELECT 1 FROM tb_account WHERE username = v_row.column3 AND deleted_at IS NULL) THEN", " RAISE EXCEPTION 'CSV 第 % 行用户名已存在: %', v_row.column1, v_row.column3;", " END IF;", " IF EXISTS (SELECT 1 FROM tb_account WHERE phone = v_row.column4 AND deleted_at IS NULL) THEN", " RAISE EXCEPTION 'CSV 第 % 行手机号已存在: %', v_row.column1, v_row.column4;", " END IF;", " END LOOP;", "END;", "$shop_import_guard$;", "", ] def _row_sql(row: ShopImportRow, plan: ShopImportPlan) -> list[str]: config = plan.config parent_expr = "NULL" if not row.parent_shop_code else f"(SELECT id FROM tb_shop WHERE shop_code = {_sql_str(row.parent_shop_code)} AND deleted_at IS NULL)" password_hash = initial_password_hash(row.phone) return [ f"-- CSV 第 {row.line_no} 行: {_sql_str(row.shop_name)}", "DO $shop_import$", "DECLARE", " v_shop_id bigint;", " v_account_id bigint;", " v_event_id bigint;", " v_parent_id bigint;", " v_operator_name varchar;", "BEGIN", f" SELECT username INTO v_operator_name FROM tb_account WHERE id = {config.operator_id};", " INSERT INTO tb_shop (creator, updater, shop_name, shop_code, parent_id, business_owner_account_id, level, contact_name, contact_phone, province, city, district, address, status, client_login_disabled, created_at, updated_at)", " VALUES (" f"{config.operator_id}, {config.operator_id}, {_sql_str(row.shop_name)}, {_sql_str(row.shop_code)}, {parent_expr}, " f"{config.salesperson_account_ids[row.salesperson]}, {row.level}, {_sql_str(row.agent_name)}, {_sql_str(row.phone)}, '', '', '', '', 1, FALSE, NOW(), NOW())", " RETURNING id, parent_id INTO v_shop_id, v_parent_id;", "", " INSERT INTO tb_account (creator, updater, username, phone, password, user_type, shop_id, is_primary, status, created_at, updated_at)", " VALUES (" f"{config.operator_id}, {config.operator_id}, {_sql_str(row.username)}, {_sql_str(row.phone)}, {_sql_str(password_hash)}, 3, v_shop_id, TRUE, 1, NOW(), NOW())", " RETURNING id INTO v_account_id;", "", " INSERT INTO tb_account_role (account_id, role_id, status, creator, updater, created_at, updated_at)", f" VALUES (v_account_id, {config.default_role_id}, 1, {config.operator_id}, {config.operator_id}, NOW(), NOW());", "", " INSERT INTO tb_shop_role (shop_id, role_id, status, creator, updater, created_at, updated_at)", f" VALUES (v_shop_id, {config.default_role_id}, 1, {config.operator_id}, {config.operator_id}, NOW(), NOW());", "", " INSERT INTO tb_agent_wallet (shop_id, wallet_type, balance, frozen_balance, credit_enabled, credit_limit, currency, status, version, shop_id_tag)", " SELECT v_shop_id, 'main', 0, 0, default_credit_enabled, default_credit_limit, 'CNY', 1, 0, v_shop_id", f" FROM tb_role WHERE id = {config.default_role_id} AND deleted_at IS NULL;", "", " INSERT INTO tb_agent_wallet (shop_id, wallet_type, balance, frozen_balance, credit_enabled, credit_limit, currency, status, version, shop_id_tag)", " VALUES (v_shop_id, 'commission', 0, 0, FALSE, 0, 'CNY', 1, 0, v_shop_id);", *_audit_sql(row, plan), "END;", "$shop_import$;", "", ] def _audit_sql(row: ShopImportRow, plan: ShopImportPlan) -> list[str]: config = plan.config owner_id = config.salesperson_account_ids[row.salesperson] created_event_id, created_hash = _audit_ids(config.batch_id, row.line_no, "shop.create") owner_event_id, owner_hash = _audit_ids(config.batch_id, row.line_no, "shop.update_business_owner") identity = ( "jsonb_build_object('id', v_shop_id, 'shop_code', " + _sql_str(row.shop_code) + ", 'shop_name', " + _sql_str(row.shop_name) + ", " "'parent_id', v_parent_id, 'level', " + str(row.level) + ")" ) after = ( "jsonb_build_object('shop_name', " + _sql_str(row.shop_name) + ", 'contact_name', " + _sql_str(row.agent_name) + ", " "'contact_phone', " + _sql_str(row.phone) + ", 'province', '', 'city', '', 'district', '', 'address', '', " "'shop_code', " + _sql_str(row.shop_code) + ", 'parent_id', v_parent_id, 'level', " + str(row.level) + ")" ) parent_resource = [ " IF v_parent_id IS NOT NULL THEN", " INSERT INTO tb_audit_event_resource (audit_event_id, resource_type, resource_id, resource_key, display_name, relation, role, identity_snapshot, before_data, after_data, subject_visibility, subject_summary, subject_data, sort_order)", " SELECT v_event_id, 'shop', id::text, id::text, shop_name, 'reference', 'shop_parent',", " jsonb_build_object('id', id, 'shop_code', shop_code, 'shop_name', shop_name, 'parent_id', parent_id, 'level', level),", " '{}'::jsonb, '{}'::jsonb, 'internal_only', '', '{}'::jsonb, 0", " FROM tb_shop WHERE id = v_parent_id;", " END IF;", ] return [ "", " INSERT INTO tb_audit_event (event_id, occurred_at, category, action_code, action_name, summary, actor_kind, actor_id, actor_name, source, scope_type, result, risk_level, metadata, content_hash)", " VALUES (" f"{_sql_str(created_event_id)}, NOW(), 'business', 'shop.create', '创建店铺', '创建店铺', 'account', '{config.operator_id}', v_operator_name, " f"'admin_api', 'platform', 'success', 'normal', jsonb_build_object('batch_id', {_sql_str(config.batch_id)}, 'csv_line_no', {row.line_no}), {_sql_str(created_hash)})", " RETURNING id INTO v_event_id;", " INSERT INTO tb_audit_event_resource (audit_event_id, resource_type, resource_id, resource_key, display_name, relation, role, identity_snapshot, before_data, after_data, subject_visibility, subject_summary, subject_data, sort_order)", " VALUES (v_event_id, 'shop', v_shop_id::text, v_shop_id::text, " + _sql_str(row.shop_name) + ", 'primary', 'shop_target', " + identity + ", '{}'::jsonb, " + after + ", 'internal_only', '', '{}'::jsonb, 0);", *parent_resource, "", " INSERT INTO tb_audit_event (event_id, occurred_at, category, action_code, action_name, summary, actor_kind, actor_id, actor_name, source, scope_type, result, risk_level, metadata, content_hash)", " VALUES (" f"{_sql_str(owner_event_id)}, NOW(), 'business', 'shop.update_business_owner', '更新店铺业务员归属', '设置店铺业务员归属', 'account', '{config.operator_id}', v_operator_name, " f"'admin_api', 'platform', 'success', 'normal', jsonb_build_object('batch_id', {_sql_str(config.batch_id)}, 'csv_line_no', {row.line_no}), {_sql_str(owner_hash)})", " RETURNING id INTO v_event_id;", " INSERT INTO tb_audit_event_resource (audit_event_id, resource_type, resource_id, resource_key, display_name, relation, role, identity_snapshot, before_data, after_data, subject_visibility, subject_summary, subject_data, sort_order)", " VALUES (v_event_id, 'shop', v_shop_id::text, v_shop_id::text, " + _sql_str(row.shop_name) + ", 'primary', 'shop_target', " + identity + ", '{}'::jsonb, " + f"jsonb_build_object('business_owner_account_id', {owner_id})" + ", 'subject_result', '店铺业务员归属已设置', '{}'::jsonb, 0);", *parent_resource, " INSERT INTO tb_audit_event_resource (audit_event_id, resource_type, resource_id, resource_key, display_name, relation, role, identity_snapshot, before_data, after_data, subject_visibility, subject_summary, subject_data, sort_order)", " SELECT v_event_id, 'account', id::text, id::text, username, 'reference', 'shop_business_owner',", " jsonb_build_object('id', id, 'username', username, 'phone', phone, 'user_type', user_type, 'shop_id', shop_id, 'enterprise_id', enterprise_id, 'wecom_userid', wecom_userid, 'wecom_name', wecom_name),", " jsonb_build_object('assigned', FALSE), jsonb_build_object('assigned', TRUE), 'internal_only', '', '{}'::jsonb, 1", f" FROM tb_account WHERE id = {owner_id};", ] def _audit_ids(batch_id: str, line_no: int, action: str) -> tuple[str, str]: seed = f"{batch_id}:{line_no}:{action}" digest = sha256(seed.encode()).hexdigest() return f"evt_{digest[:32]}", digest def _sql_str(value: str) -> str: return "'" + value.replace("'", "''") + "'"