#!/usr/bin/env python3 # -*- coding: utf-8 -*- """B01 展示字段批次：安全演示数据工具（预检 / 预览 / 导入 / 定向清理 / 核对）。 设计目标（本版重写归属逻辑） --------------------------- * **先预检、后写入**：任何一次导入在**业务写入之前**先完成全部冲突预检；只要出现 无法证明属本批的冲突（固定手机号被非本批账号占用、同名企业/岗位已被非本批记录占用、 账本丢失/资源丢失/身份变化），立即以**非零退出码**失败，绝不自动认领、覆盖、补角色、 置认证或删除。 * **可靠的批次归属账本**：仅工具使用的辅助表 ``b01_demo_ledger``（**不进入** Flyway 迁移或 应用启动流程）。账本只登记**本工具本次实际创建**的记录，逐条记录：批次 ID、目标库实例 身份、资源类型、**实际创建的主键 ID**、稳定演示条目键（demo_key）、确认资源身份所需的 指纹。**禁止**凭手机号/企业名/岗位标题/描述推断归属。 * **写入与归属登记同事务**：业务记录创建与账本写入在**同一事务**完成，失败整体回滚。 账本表的 DDL（``CREATE TABLE IF NOT EXISTS``）在业务事务**之前**单独执行——MySQL DDL 会隐式提交且不可回滚，故必须与业务事务分开说明与执行。 * **重复导入先校验归属**：账本中资源存在且身份一致 → 跳过（幂等）；资源丢失 / ID 指向 异常 / 身份变化 / 账本不完整 → 明确报冲突，不静默接管、不自动修复。 * **清理只用账本精确 ID**：删除前核验资源身份与**依赖**（通过 information_schema 枚举外键 引用方）；发现本批之外的依赖、角色或业务关联 → 拒绝清理并给出原因；不级联清除他人数据， 不关闭外键检查。全程在事务内以 ``SELECT ... FOR UPDATE`` 加行锁，避免「检查—删除」竞态。 * **预览与执行共用同一套计划**：默认 dry-run、零写入（**不创建**任何辅助表）；导入/清理预览 与执行都基于同一 ``build_plan`` / ``build_cleanup_plan``，执行前针对事务内最新状态**重新 校验**；verify 按账本精确范围统计，不使用 ``LIKE`` 模糊统计；明确区分「没有本批数据」与 「账本缺失，无法证明归属」。 * 保留既有隔离目标限制（回环地址 + 显式白名单三元组 + 线上地址阻断）、显式 ``--apply``、 虚构演示标识；**不伪造**评价、评分、浏览量、真实投递；不关闭任何认证/安全策略。 用法 ---- python docs/tools/b01_demo_display_data.py # dry-run 预览（零写入） python docs/tools/b01_demo_display_data.py --apply # 首次/重复导入 python docs/tools/b01_demo_display_data.py --cleanup # 清理预览（零写入） python docs/tools/b01_demo_display_data.py --cleanup --apply # 定向清理 python docs/tools/b01_demo_display_data.py --verify # 按账本精确核对 退出码：0=成功；2=参数/目标拒绝；3=归属冲突/无法证明归属而拒绝；1=意外错误。 """ import argparse import json import sys import pymysql BATCH_ID = "b01-demo-2026" DEMO_MARK = "[演示数据·非真实招聘]" LEDGER_TABLE = "b01_demo_ledger" # 显式白名单：只有精确匹配的三元组才允许操作。端口 3308 为本批隔离容器固定映射端口。 # 测试隔离库 mydreamjob_b01_demo_test 与演示库同属本批隔离范围，一并显式登记。 ALLOWED_TARGETS = { ("127.0.0.1", 3308, "mydreamjob_b01_demo"), ("127.0.0.1", 3308, "mydreamjob_b01_demo_test"), } # 已知线上/生产地址特征：命中即拒绝。 BLOCKED_HOST_MARKERS = ( "prod", "production", "online", "rds.aliyuncs.com", "mydreamjob.com", "zhilai.com", "aliyuncs", "amazonaws", ) LOOPBACK = {"127.0.0.1", "localhost", "::1"} COMPANIES = [ dict(phone="19901000001", name="星河智造科技（演示）", industry="智能制造", company_size="500-999人", website="https://demo.example/starship", description=DEMO_MARK + " 虚构智能制造企业，用于展示岗位字段，不涉及真实招聘。"), dict(phone="19901000002", name="云栖数据服务（演示）", industry="大数据", company_size="100-499人", website="https://demo.example/cloudnest", description=DEMO_MARK + " 虚构数据服务企业，用于展示岗位字段，不涉及真实招聘。"), dict(phone="19901000003", name="沐光文化传媒（演示）", industry="文化传媒", company_size="20-99人", website="https://demo.example/sunlit", description=DEMO_MARK + " 虚构文化传媒企业，用于展示岗位字段，不涉及真实招聘。"), ] JOBS = [ # 星河智造科技（演示） dict(phone="19901000001", title="工业机器人算法实习生", city="上海", employment_type="INTERNSHIP", salary_min=4000, salary_max=6000, experience="应届生", education="本科", benefits=["导师带教", "免费班车", "实习转正"], salary_months=12, work_location="上海市浦东新区", description=DEMO_MARK + " 参与机器人运动控制算法开发。", requirements="自动化/计算机相关专业。"), dict(phone="19901000001", title="嵌入式软件工程师", city="上海", employment_type="FULL_TIME", salary_min=20000, salary_max=35000, experience="3-5年", education="本科", benefits=["五险一金", "弹性工作", "年度体检", "股票期权"], salary_months=14, work_location="上海市闵行区", description=DEMO_MARK + " 负责机器人控制器嵌入式软件开发。", requirements="熟悉 C/C++ 与 RTOS。"), dict(phone="19901000001", title="高级机械结构工程师", city="苏州", employment_type="FULL_TIME", salary_min=30000, salary_max=50000, experience="5-10年", education="硕士", benefits=["五险一金", "补充医疗", "带薪年假"], salary_months=15, work_location="江苏省苏州市工业园区", description=DEMO_MARK + " 负责精密机械结构设计与验证。", requirements="机械工程硕士，5 年以上经验。"), dict(phone="19901000001", title="供应链计划专员", city="上海", employment_type="FULL_TIME", salary_min=12000, salary_max=18000, experience="1-3年", education="大专", benefits=["五险一金", "餐补", "节日福利"], salary_months=13, work_location="上海市嘉定区", description=DEMO_MARK + " 负责物料计划与库存管理。", requirements="供应链或物流相关背景。"), # 云栖数据服务（演示） dict(phone="19901000002", title="数据平台后端工程师", city="杭州", employment_type="FULL_TIME", salary_min=25000, salary_max=45000, experience="3-5年", education="本科", benefits=["五险一金", "弹性工作", "免费三餐", "技术分享"], salary_months=16, work_location="浙江省杭州市余杭区", description=DEMO_MARK + " 负责数据平台服务端研发。", requirements="扎实的 Java 与分布式基础。"), dict(phone="19901000002", title="大数据开发工程师", city="杭州", employment_type="FULL_TIME", salary_min=22000, salary_max=40000, experience="3-5年", education="本科", benefits=["五险一金", "年度体检", "带薪年假"], salary_months=14, work_location="浙江省杭州市滨江区", description=DEMO_MARK + " 负责离线与实时数仓开发。", requirements="熟悉 Spark/Flink。"), dict(phone="19901000002", title="数据分析师（校招）", city="杭州", employment_type="FULL_TIME", salary_min=15000, salary_max=22000, experience="应届生", education="本科", benefits=["五险一金", "租房补贴", "培训体系"], salary_months=13, work_location="浙江省杭州市余杭区", description=DEMO_MARK + " 负责业务数据分析与报表建设。", requirements="统计学/数学相关专业。"), dict(phone="19901000002", title="SRE 运维工程师", city="远程", employment_type="FULL_TIME", salary_min=28000, salary_max=48000, experience="5-10年", education="不限", benefits=["五险一金", "远程办公", "弹性工作", "年度体检"], salary_months=15, work_location="全国远程", description=DEMO_MARK + " 负责平台稳定性与可观测性建设。", requirements="熟悉 Kubernetes 与监控体系。"), # 沐光文化传媒（演示） dict(phone="19901000003", title="短视频内容运营", city="成都", employment_type="FULL_TIME", salary_min=8000, salary_max=13000, experience="1-3年", education="大专", benefits=["五险一金", "下午茶", "节日福利"], salary_months=12, work_location="四川省成都市武侯区", description=DEMO_MARK + " 负责短视频账号内容策划与运营。", requirements="有内容运营经验优先。"), dict(phone="19901000003", title="视觉设计师", city="成都", employment_type="FULL_TIME", salary_min=12000, salary_max=20000, experience="3-5年", education="本科", benefits=["五险一金", "弹性工作", "作品集补贴"], salary_months=13, work_location="四川省成都市高新区", description=DEMO_MARK + " 负责品牌视觉与活动设计。", requirements="设计相关专业，作品集优秀。"), dict(phone="19901000003", title="市场活动策划", city="成都", employment_type="PART_TIME", salary_min=6000, salary_max=9000, experience="1-3年", education="不限", benefits=["五险一金", "项目奖金"], salary_months=12, work_location="四川省成都市锦江区", description=DEMO_MARK + " 负责线下市场活动策划与执行。", requirements="具备活动执行经验。"), dict(phone="19901000003", title="客户成功经理", city="重庆", employment_type="FULL_TIME", salary_min=15000, salary_max=25000, experience="3-5年", education="本科", benefits=["五险一金", "补充医疗", "带薪年假", "团队旅游"], salary_months=14, work_location="重庆市渝北区", description=DEMO_MARK + " 负责重点客户关系维护与续约。", requirements="具备 B 端客户服务经验。"), ] # 稳定演示条目键与身份指纹：账本据此判定「本批资源身份」是否与当前库内记录一致。 def user_demo_key(phone): return "user:" + phone def company_demo_key(phone): return "company:" + phone def job_demo_key(phone, title): return "job:%s|%s" % (phone, title) def user_fingerprint(phone): return "phone=%s" % phone def company_fingerprint(phone, name): return "owner_phone=%s;name=%s" % (phone, name) def job_fingerprint(phone, title): return "company_phone=%s;title=%s" % (phone, title) LEDGER_DDL = """ CREATE TABLE IF NOT EXISTS %s ( id BIGINT AUTO_INCREMENT PRIMARY KEY, batch_id VARCHAR(64) NOT NULL, target_identity VARCHAR(160) NOT NULL, resource_type VARCHAR(16) NOT NULL, resource_id BIGINT NOT NULL, demo_key VARCHAR(191) NOT NULL, identity_fingerprint VARCHAR(255) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_ledger_target_key (target_identity, demo_key), UNIQUE KEY uk_ledger_target_res (target_identity, resource_type, resource_id), KEY idx_ledger_batch (batch_id, target_identity) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ % LEDGER_TABLE # ── 连接与目标限制 ────────────────────────────────────────────────── def assert_safe_target(host, port, database): """显式白名单 + 线上地址阻断。任何不满足都直接终止，不降级。""" for marker in BLOCKED_HOST_MARKERS: if marker in host.lower() or marker in database.lower(): raise SystemExit(f"[REFUSED] 目标疑似线上地址，已拒绝：host={host} database={database}") if host not in LOOPBACK: raise SystemExit(f"[REFUSED] 目标 host 非回环地址：{host}") if (host, port, database) not in ALLOWED_TARGETS: raise SystemExit( f"[REFUSED] 目标不在显式白名单内：{(host, port, database)}；" f"允许：{sorted(ALLOWED_TARGETS)}") if "b01" not in database: raise SystemExit(f"[REFUSED] 数据库名缺少本批标识 b01：{database}") def connect(args): assert_safe_target(args.host, args.port, args.database) return pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database=args.database, charset="utf8mb4", autocommit=False) # ── 查询助手 ──────────────────────────────────────────────────────── def query(cur, sql, params=None): cur.execute(sql, params if params else None) cols = [d[0] for d in cur.description] return [dict(zip(cols, row)) for row in cur.fetchall()] def query_one(cur, sql, params=None): rows = query(cur, sql, params) return rows[0] if rows else None def scalar(cur, sql, params=None): row = query_one(cur, sql, params) if not row: return None return list(row.values())[0] def table_exists(cur, name): n = scalar(cur, "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema=DATABASE() AND table_name=%s", (name,)) return int(n or 0) > 0 REQUIRED_BUSINESS_TABLES = ("users", "user_roles", "companies", "jobs") def assert_business_schema(cur): missing = [t for t in REQUIRED_BUSINESS_TABLES if not table_exists(cur, t)] if missing: raise SystemExit("[REFUSED] 目标库缺少必需业务表 %s，拒绝操作" % missing) def target_identity(cur, args): """目标库实例身份：优先 server_uuid（可区分不同实例），退化到 host:port。""" uuid = None try: uuid = scalar(cur, "SELECT @@server_uuid") except Exception: uuid = None if not uuid: uuid = "%s:%s" % (args.host, args.port) return "%s/%s" % (uuid, args.database) # ── 账本读写 ──────────────────────────────────────────────────────── def read_ledger(cur, identity, for_update=False): """返回本批本目标的账本行；账本表不存在返回 None（区别于空列表）。""" if not table_exists(cur, LEDGER_TABLE): return None sql = ("SELECT id, resource_type, resource_id, demo_key, identity_fingerprint " "FROM %s WHERE batch_id=%%s AND target_identity=%%s" % LEDGER_TABLE) if for_update: sql += " FOR UPDATE" rows = query(cur, sql, (BATCH_ID, identity)) for r in rows: r["resource_id"] = int(r["resource_id"]) return rows def insert_ledger(cur, identity, resource_type, resource_id, demo_key, fingerprint): cur.execute( "INSERT INTO %s(batch_id, target_identity, resource_type, resource_id, demo_key, " "identity_fingerprint) VALUES(%%s,%%s,%%s,%%s,%%s,%%s)" % LEDGER_TABLE, (BATCH_ID, identity, resource_type, resource_id, demo_key, fingerprint)) # ── 计划构建（预检） ──────────────────────────────────────────────── def validate_ledger_shape(ledger): """账本键、类型和指纹必须属于当前定义；不能用损坏的账本证明归属。""" expected = {} for c in COMPANIES: expected[user_demo_key(c['phone'])] = ('USER', user_fingerprint(c['phone'])) expected[company_demo_key(c['phone'])] = ('COMPANY', company_fingerprint(c['phone'], c['name'])) for j in JOBS: expected[job_demo_key(j['phone'], j['title'])] = ('JOB', job_fingerprint(j['phone'], j['title'])) conflicts = [] seen = set() for row in ledger or []: key = row['demo_key'] if (key in seen or expected.get(key) != (row['resource_type'], row['identity_fingerprint']) or row['resource_id'] <= 0): conflicts.append('账本键/类型/指纹/ID 异常：%s，拒绝操作' % key) seen.add(key) return conflicts def build_plan(cur, identity, ledger): """只读地计算导入计划。返回 (actions, conflicts)。 actions 中每条：kind = create_*/keep_*；携带 type/demo_key/label 与已有 id（如有）。 conflicts 非空表示无法证明本批归属，调用方必须拒绝写入。 """ actions, conflicts = [], validate_ledger_shape(ledger) if conflicts: return actions, conflicts ledger_by_key = {r["demo_key"]: r for r in (ledger or [])} used_keys = set() user_ids, company_ids = {}, {} # 用户 for c in COMPANIES: phone = c["phone"] key = user_demo_key(phone) used_keys.add(key) existing = query_one(cur, "SELECT id FROM users WHERE phone=%s", (phone,)) lr = ledger_by_key.get(key) if existing is None: if lr is not None: conflicts.append("账本记录本批用户 %s(id=%s) 已丢失：拒绝静默重建或接管" % (phone, lr["resource_id"])) user_ids[phone] = None else: actions.append(dict(kind="create_user", type="USER", demo_key=key, phone=phone, label=phone, id=None)) user_ids[phone] = None continue if lr is None: conflicts.append("手机号 %s 已被非本批账号占用(users.id=%s)且账本无本批归属记录：" "拒绝认领/覆盖/补角色" % (phone, existing["id"])) user_ids[phone] = existing["id"] continue if int(lr["resource_id"]) != int(existing["id"]) \ or lr["identity_fingerprint"] != user_fingerprint(phone): conflicts.append("手机号 %s 账本与现有账号不一致(账本 id=%s vs 实际 id=%s)：拒绝接管" % (phone, lr["resource_id"], existing["id"])) user_ids[phone] = existing["id"] continue actions.append(dict(kind="keep_user", type="USER", demo_key=key, phone=phone, label=phone, id=existing["id"])) user_ids[phone] = existing["id"] # 企业 for c in COMPANIES: phone, name = c["phone"], c["name"] key = company_demo_key(phone) used_keys.add(key) uid = user_ids.get(phone) label = "%s / %s" % (phone, name) lr = ledger_by_key.get(key) by_owner = (query_one(cur, "SELECT id, name FROM companies WHERE owner_user_id=%s", (uid,)) if uid is not None else None) by_name = query_one(cur, "SELECT id, owner_user_id FROM companies WHERE name=%s", (name,)) if uid is None: # 演示账号即将新建 → 企业必为新建；但必须确保没有同名企业被他方占用 if lr is not None: conflicts.append("账本记录本批企业 %s(id=%s) 已丢失：拒绝静默重建" % (name, lr["resource_id"])) company_ids[phone] = None elif by_name is not None: conflicts.append("企业名 %s 已被非本批企业占用(companies.id=%s, owner_user_id=%s)：" "拒绝覆盖/认领" % (name, by_name["id"], by_name["owner_user_id"])) company_ids[phone] = None else: actions.append(dict(kind="create_company", type="COMPANY", demo_key=key, phone=phone, name=name, label=label, id=None)) company_ids[phone] = None continue if by_owner is None: if lr is not None: conflicts.append("账本记录本批企业 %s(id=%s, owner=%s) 已丢失：拒绝静默重建" % (name, lr["resource_id"], phone)) company_ids[phone] = None continue if by_name is not None: conflicts.append("企业名 %s 已被非本批企业占用(companies.id=%s, owner_user_id=%s)：" "拒绝覆盖/认领" % (name, by_name["id"], by_name["owner_user_id"])) company_ids[phone] = None continue actions.append(dict(kind="create_company", type="COMPANY", demo_key=key, phone=phone, name=name, label=label, id=None)) company_ids[phone] = None continue # 演示账号名下已有企业：必须能由账本证明归属，且不能另有同名企业 ok = (lr is not None and int(lr["resource_id"]) == int(by_owner["id"]) and lr["identity_fingerprint"] == company_fingerprint(phone, by_owner["name"])) if not ok: conflicts.append("演示账号 %s 名下已有企业(companies.id=%s)但无本批账本归属或身份不一致：" "拒绝覆盖企业资料/置认证" % (phone, by_owner["id"])) company_ids[phone] = by_owner["id"] continue if by_name is not None and int(by_name["id"]) != int(by_owner["id"]): conflicts.append("企业名 %s 存在同名重复记录(companies.id=%s 非本批)：拒绝继续" % (name, by_name["id"])) company_ids[phone] = by_owner["id"] continue actions.append(dict(kind="keep_company", type="COMPANY", demo_key=key, phone=phone, name=name, label=label, id=by_owner["id"])) company_ids[phone] = by_owner["id"] # 岗位 for j in JOBS: phone, title = j["phone"], j["title"] key = job_demo_key(phone, title) used_keys.add(key) cid = company_ids.get(phone) label = "%s / %s" % (phone, title) if cid is None: lr = ledger_by_key.get(key) if lr is not None: conflicts.append("账本记录本批岗位 %s(id=%s) 已丢失：拒绝静默重建" % (label, lr["resource_id"])) else: actions.append(dict(kind="create_job", type="JOB", demo_key=key, phone=phone, title=title, label=label, id=None)) continue existing = query_one(cur, "SELECT id FROM jobs WHERE company_id=%s AND title=%s", (cid, title)) lr = ledger_by_key.get(key) if existing is None: if lr is not None: conflicts.append("账本记录本批岗位 %s(id=%s) 已丢失：拒绝静默重建" % (label, lr["resource_id"])) else: actions.append(dict(kind="create_job", type="JOB", demo_key=key, phone=phone, title=title, label=label, id=None)) continue ok = (lr is not None and int(lr["resource_id"]) == int(existing["id"]) and lr["identity_fingerprint"] == job_fingerprint(phone, title)) if not ok: conflicts.append("本批企业 %s 下已存在同名岗位 '%s'(jobs.id=%s)但无本批账本归属：" "拒绝认领/覆盖" % (phone, title, existing["id"])) continue actions.append(dict(kind="keep_job", type="JOB", demo_key=key, phone=phone, title=title, label=label, id=existing["id"])) # 账本中存在当前批次定义之外的记录 → 拒绝（不自动处理、不猜测归属） for r in (ledger or []): if r["demo_key"] not in used_keys: conflicts.append("账本含当前批次定义之外的记录(%s/%s id=%s)：拒绝自动处理" % (r["resource_type"], r["demo_key"], r["resource_id"])) return actions, conflicts # ── 导入执行 ──────────────────────────────────────────────────────── def _benefits_json(benefits): return json.dumps(benefits, ensure_ascii=False, separators=(",", ":")) def apply_import(conn, cur, identity): """执行导入：先预检（零写入），无冲突后再建账本表 + 业务写入/账本登记同事务。 返回 (created, conflicts)。conflicts 非空时不写入任何业务数据、也不创建账本表。 """ # 1) 预检：只读，不创建任何表、不写任何行。 ledger = read_ledger(cur, identity) _, conflicts = build_plan(cur, identity, ledger) if conflicts: conn.rollback() return None, conflicts # 2) 账本表 DDL（MySQL DDL 隐式提交、不可回滚，故独立于业务事务）。 cur.execute(LEDGER_DDL) conn.commit() # 3) 业务事务：针对最新状态重新校验后写入。 try: ledger = read_ledger(cur, identity) actions, conflicts = build_plan(cur, identity, ledger) if conflicts: conn.rollback() return None, conflicts comp_by_phone = {c["phone"]: c for c in COMPANIES} job_by_key = {job_demo_key(j["phone"], j["title"]): j for j in JOBS} created = dict(users=0, companies=0, jobs=0) user_ids, company_ids = {}, {} for a in actions: if a["kind"] == "keep_user": user_ids[a["phone"]] = a["id"] elif a["kind"] == "create_user": cur.execute("INSERT INTO users(phone, nickname, role, is_active, is_verified) " "VALUES(%s,%s,'COMPANY',1,1)", (a["phone"], a["phone"] + "（演示账号）")) uid = cur.lastrowid cur.execute("INSERT INTO user_roles(user_id, role) VALUES(%s,'COMPANY')", (uid,)) insert_ledger(cur, identity, "USER", uid, a["demo_key"], user_fingerprint(a["phone"])) user_ids[a["phone"]] = uid created["users"] += 1 elif a["kind"] == "keep_company": company_ids[a["phone"]] = a["id"] elif a["kind"] == "create_company": src = comp_by_phone[a["phone"]] uid = user_ids[a["phone"]] cur.execute("INSERT INTO companies(owner_user_id, name, logo_url, description, " "industry, company_size, website, address, verified) " "VALUES(%s,%s,NULL,%s,%s,%s,%s,NULL,1)", (uid, src["name"], src["description"], src["industry"], src["company_size"], src["website"])) cid = cur.lastrowid company_ids[a["phone"]] = cid insert_ledger(cur, identity, "COMPANY", cid, a["demo_key"], company_fingerprint(a["phone"], src["name"])) created["companies"] += 1 elif a["kind"] == "keep_job": continue elif a["kind"] == "create_job": j = job_by_key[a["demo_key"]] cid = company_ids[a["phone"]] cur.execute("INSERT INTO jobs(company_id, title, description, requirements, city, " "employment_type, salary_min, salary_max, experience, education, " "benefits_json, salary_months, work_location, status, published_at) " "VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'PUBLISHED',NOW())", (cid, j["title"], j["description"], j["requirements"], j["city"], j["employment_type"], j["salary_min"], j["salary_max"], j["experience"], j["education"], _benefits_json(j["benefits"]), j["salary_months"], j["work_location"])) jid = cur.lastrowid insert_ledger(cur, identity, "JOB", jid, a["demo_key"], job_fingerprint(a["phone"], j["title"])) created["jobs"] += 1 conn.commit() return created, [] except Exception: conn.rollback() raise # ── 清理计划与执行 ────────────────────────────────────────────────── def _dependent_columns(cur, table): return query(cur, "SELECT TABLE_NAME AS t, COLUMN_NAME AS c FROM information_schema.KEY_COLUMN_USAGE " "WHERE REFERENCED_TABLE_SCHEMA=DATABASE() AND REFERENCED_TABLE_NAME=%s " "AND REFERENCED_COLUMN_NAME='id' ORDER BY TABLE_NAME", (table,)) def _count_refs(cur, table, column, rid): return int(scalar(cur, "SELECT COUNT(*) FROM `%s` WHERE `%s`=%%s" % (table, column), (rid,)) or 0) def build_cleanup_plan(cur, identity, ledger, for_update=False): """只读（或加锁）地计算清理计划。返回 (plan, conflicts)。 ledger 为 None → 账本缺失，无法证明归属。ledger 为空 → 无本批数据。 """ if ledger is None: return None, ["账本表不存在：无法证明本批归属，拒绝清理"] plan = dict(jobs=[], companies=[], users=[], ledger_ids=[]) conflicts = [] if not ledger: return plan, conflicts # 已清理或无本批数据 → 安全无操作 conflicts.extend(validate_ledger_shape(ledger)) if conflicts: return plan, conflicts ledger_by_key = {r['demo_key']: r for r in ledger} job_rows = [r for r in ledger if r["resource_type"] == "JOB"] comp_rows = [r for r in ledger if r["resource_type"] == "COMPANY"] user_rows = [r for r in ledger if r["resource_type"] == "USER"] ledger_company_ids = {r["resource_id"] for r in comp_rows} ledger_job_ids = {r["resource_id"] for r in job_rows} # 账本完整性：COMPANY 必须有对应 USER，JOB 必须有对应 COMPANY for r in comp_rows: owner_phone = r["demo_key"].split(":", 1)[1] if not any(u["demo_key"] == user_demo_key(owner_phone) for u in user_rows): conflicts.append("账本不完整：企业 %s 缺少归属用户账本记录" % r["demo_key"]) for r in job_rows: company_key = r["demo_key"].split("|", 1)[0].replace("job:", "company:") if not any(c["demo_key"] == company_key for c in comp_rows): conflicts.append("账本不完整：岗位 %s 缺少归属企业账本记录" % r["demo_key"]) # 用户 for r in user_rows: uid = r["resource_id"] phone = r["demo_key"].split(":", 1)[1] row = query_one(cur, "SELECT id, phone FROM users WHERE id=%s" + (" FOR UPDATE" if for_update else ""), (uid,)) if row is None: conflicts.append("账本记录用户 %s(id=%s) 已丢失：无法确认清理范围，拒绝" % (phone, uid)) continue if row['phone'] != phone or r["identity_fingerprint"] != user_fingerprint(row['phone']): conflicts.append("账本用户 %s(id=%s) 身份指纹异常：拒绝清理" % (phone, uid)) continue # 允许的依赖：名下本批企业 + 仅 COMPANY 角色；其余依赖一律拒绝 for dep in _dependent_columns(cur, "users"): t, c = dep["t"], dep["c"] n = _count_refs(cur, t, c, uid) if n == 0: continue if t == "companies" and c == "owner_user_id": owned = [int(x["id"]) for x in query( cur, "SELECT id FROM companies WHERE owner_user_id=%s", (uid,))] extra = [cid for cid in owned if cid not in ledger_company_ids] if extra: conflicts.append("用户 %s 名下存在非本批企业 %s：拒绝清理" % (phone, extra)) elif t == "user_roles" and c == "user_id": roles = [x["role"] for x in query( cur, "SELECT role FROM user_roles WHERE user_id=%s", (uid,))] extra = [x for x in roles if x != "COMPANY"] if extra: conflicts.append("用户 %s 存在非本批角色 %s：拒绝清理" % (phone, extra)) else: conflicts.append("用户 %s 存在本批之外的依赖 %s.%s(%d 行)：拒绝清理" % (phone, t, c, n)) plan["users"].append(dict(id=uid, phone=phone, demo_key=r["demo_key"])) # 企业 for r in comp_rows: cid = r["resource_id"] phone = r["demo_key"].split(":", 1)[1] suffix = " FOR UPDATE" if for_update else "" row = query_one(cur, "SELECT c.id AS id, c.name AS name, c.owner_user_id, u.phone AS owner_phone " "FROM companies c JOIN users u ON u.id=c.owner_user_id " "WHERE c.id=%s" + suffix, (cid,)) if row is None: conflicts.append("账本记录企业 %s(id=%s) 已丢失：无法确认清理范围，拒绝" % (phone, cid)) continue parent = ledger_by_key.get(user_demo_key(phone)) if (parent is None or row['owner_user_id'] != parent['resource_id'] or row['owner_phone'] != phone or r["identity_fingerprint"] != company_fingerprint(row["owner_phone"], row["name"])): conflicts.append("账本企业 %s(id=%s) 身份指纹异常：拒绝清理" % (phone, cid)) continue for dep in _dependent_columns(cur, "companies"): t, c = dep["t"], dep["c"] n = _count_refs(cur, t, c, cid) if n == 0: continue if t == "jobs" and c == "company_id": jobs_here = [int(x["id"]) for x in query( cur, "SELECT id FROM jobs WHERE company_id=%s", (cid,))] extra = [jid for jid in jobs_here if jid not in ledger_job_ids] if extra: conflicts.append("企业 %s 下存在非本批岗位 %s：拒绝清理" % (phone, extra)) else: conflicts.append("企业 %s 存在本批之外的依赖 %s.%s(%d 行)：拒绝清理" % (phone, t, c, n)) plan["companies"].append(dict(id=cid, phone=phone, demo_key=r["demo_key"])) # 岗位 for r in job_rows: jid = r["resource_id"] suffix = " FOR UPDATE" if for_update else "" row = query_one(cur, "SELECT j.id AS id, j.title AS title, j.company_id, u.phone AS owner_phone " "FROM jobs j JOIN companies c ON c.id=j.company_id " "JOIN users u ON u.id=c.owner_user_id WHERE j.id=%s" + suffix, (jid,)) if row is None: conflicts.append("账本记录岗位 id=%s 已丢失：无法确认清理范围，拒绝" % jid) continue parent_key = r['demo_key'].split('|', 1)[0].replace('job:', 'company:', 1) parent = ledger_by_key.get(parent_key) if (parent is None or row['company_id'] != parent['resource_id'] or r["identity_fingerprint"] != job_fingerprint(row["owner_phone"], row["title"])): conflicts.append("账本岗位 id=%s 身份指纹异常：拒绝清理" % jid) continue for dep in _dependent_columns(cur, "jobs"): t, c = dep["t"], dep["c"] n = _count_refs(cur, t, c, jid) if n > 0: conflicts.append("岗位 id=%s 存在本批之外的依赖 %s.%s(%d 行)：拒绝清理" % (jid, t, c, n)) plan["jobs"].append(dict(id=jid, company_id=row['company_id'], demo_key=r["demo_key"])) plan["ledger_ids"] = [r["id"] for r in ledger] return plan, conflicts def apply_cleanup(conn, cur, identity): """执行清理：事务 + 行锁，只删除账本精确 ID。返回 (removed, conflicts)。""" try: ledger = read_ledger(cur, identity, for_update=True) plan, conflicts = build_cleanup_plan(cur, identity, ledger, for_update=True) if conflicts: conn.rollback() return None, conflicts removed = dict(jobs=0, companies=0, users=0, ledger=0) for job in plan["jobs"]: cur.execute("DELETE FROM jobs WHERE id=%s AND company_id=%s", (job["id"], job['company_id'])) if cur.rowcount != 1: raise RuntimeError('岗位删除目标变化，回滚全部清理') removed["jobs"] += cur.rowcount for comp in plan["companies"]: cur.execute("DELETE FROM companies WHERE id=%s AND owner_user_id=" "(SELECT id FROM users WHERE phone=%s)", (comp["id"], comp["phone"])) if cur.rowcount != 1: raise RuntimeError('企业删除目标变化，回滚全部清理') removed["companies"] += cur.rowcount for user in plan["users"]: cur.execute("DELETE FROM user_roles WHERE user_id=%s", (user["id"],)) cur.execute("DELETE FROM users WHERE id=%s AND phone=%s", (user["id"], user['phone'])) if cur.rowcount != 1: raise RuntimeError('用户删除目标变化，回滚全部清理') removed["users"] += cur.rowcount if plan["ledger_ids"]: fmt = ",".join(["%s"] * len(plan["ledger_ids"])) cur.execute("DELETE FROM %s WHERE id IN (%s)" % (LEDGER_TABLE, fmt), tuple(plan["ledger_ids"])) removed["ledger"] = cur.rowcount if cur.rowcount != len(plan['ledger_ids']): raise RuntimeError('账本删除范围变化，回滚全部清理') conn.commit() return removed, [] except Exception: conn.rollback() raise # ── 核对 ──────────────────────────────────────────────────────────── def verify(cur, identity): """按账本精确范围核对（不使用 LIKE 模糊统计）。""" ledger = read_ledger(cur, identity) demo_phones = [c["phone"] for c in COMPANIES] demo_names = [c["name"] for c in COMPANIES] demo_titles = [(j["phone"], j["title"]) for j in JOBS] def present_demo_users(): fmt = ",".join(["%s"] * len(demo_phones)) return [r["id"] for r in query( cur, "SELECT id FROM users WHERE phone IN (%s)" % fmt, tuple(demo_phones))] def present_demo_companies(): fmt = ",".join(["%s"] * len(demo_names)) return [r["id"] for r in query( cur, "SELECT id FROM companies WHERE name IN (%s)" % fmt, tuple(demo_names))] def present_demo_jobs(): ids = [] for phone, title in demo_titles: owner = query_one(cur, "SELECT id FROM users WHERE phone=%s", (phone,)) if not owner: continue comp = query_one(cur, "SELECT id FROM companies WHERE owner_user_id=%s", (owner["id"],)) if not comp: continue job = query_one(cur, "SELECT id FROM jobs WHERE company_id=%s AND title=%s", (comp["id"], title)) if job: ids.append(job["id"]) return ids if ledger is None: return dict(status="LEDGER_MISSING", detail="账本表不存在：无法证明本批归属；不得据此清理", demo_users_present=present_demo_users(), demo_companies_present=present_demo_companies(), demo_jobs_present=present_demo_jobs()) counts = dict(USER=0, COMPANY=0, JOB=0) for r in ledger: counts[r["resource_type"]] = counts.get(r["resource_type"], 0) + 1 alive = dict(USER=0, COMPANY=0, JOB=0) for r in ledger: if r["resource_type"] == "USER": row = query_one(cur, "SELECT id FROM users WHERE id=%s", (r["resource_id"],)) elif r["resource_type"] == "COMPANY": row = query_one(cur, "SELECT id FROM companies WHERE id=%s", (r["resource_id"],)) else: row = query_one(cur, "SELECT id FROM jobs WHERE id=%s", (r["resource_id"],)) if row is not None: alive[r["resource_type"]] += 1 if not ledger: status = "NO_BATCH_DATA" detail = "账本无本批记录" else: status = "HAS_BATCH_DATA" detail = "按账本精确范围统计" return dict(status=status, detail=detail, ledger_counts=counts, ledger_alive=alive, demo_users_present=present_demo_users(), demo_companies_present=present_demo_companies(), demo_jobs_present=present_demo_jobs()) # ── 计划打印 ──────────────────────────────────────────────────────── def print_plan(actions, counters): print("[%s] 导入预览（dry-run，未写入，共 %d 项动作）：" % (BATCH_ID, len(actions))) for a in actions: print(" %-14s %s" % (a["kind"], a["label"])) print(" 计划新建:", counters) def print_cleanup_plan(plan): print("[%s] 清理预览（dry-run，未写入）：" % BATCH_ID) print(" 将删除岗位:", [j["id"] for j in plan["jobs"]]) print(" 将删除企业:", [c["id"] for c in plan["companies"]]) print(" 将删除用户:", [u["id"] for u in plan["users"]]) print(" 将删除账本行:", len(plan["ledger_ids"])) # ── CLI ───────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="B01 展示字段批次演示数据工具（默认 dry-run）") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=3308) parser.add_argument("--database", default="mydreamjob_b01_demo") parser.add_argument("--user", default="b01demo") parser.add_argument("--password", default="b01-demo-only") parser.add_argument("--apply", action="store_true", help="显式确认执行写入") parser.add_argument("--cleanup", action="store_true", help="执行定向清理（需配合 --apply）") parser.add_argument("--verify", action="store_true", help="按账本精确核对") args = parser.parse_args() conn = connect(args) try: with conn.cursor() as cur: assert_business_schema(cur) identity = target_identity(cur, args) if args.verify: print("[%s] verify identity=%s:" % (BATCH_ID, identity), json.dumps(verify(cur, identity), ensure_ascii=False)) return 0 if args.cleanup: ledger = read_ledger(cur, identity, for_update=args.apply) plan, conflicts = build_cleanup_plan(cur, identity, ledger, for_update=args.apply) if conflicts: print("[%s] 清理拒绝：" % BATCH_ID) for c in conflicts: print(" -", c) conn.rollback() return 3 if not args.apply: print_cleanup_plan(plan) conn.rollback() return 0 removed, conflicts = apply_cleanup(conn, cur, identity) if conflicts: print("[%s] 清理拒绝：" % BATCH_ID) for c in conflicts: print(" -", c) return 3 print("[%s] cleanup applied:" % BATCH_ID, removed) return 0 ledger = read_ledger(cur, identity) actions, conflicts = build_plan(cur, identity, ledger) if conflicts: print("[%s] 导入预检失败，已拒绝任何写入：" % BATCH_ID) for c in conflicts: print(" -", c) conn.rollback() return 3 if not args.apply: counters = dict( users=sum(1 for a in actions if a["kind"] == "create_user"), companies=sum(1 for a in actions if a["kind"] == "create_company"), jobs=sum(1 for a in actions if a["kind"] == "create_job")) print_plan(actions, counters) conn.rollback() return 0 # 执行前针对事务内最新状态重新校验（apply_import 内部重新 build_plan） created, conflicts = apply_import(conn, cur, identity) if conflicts: print("[%s] 导入预检失败，已拒绝任何写入：" % BATCH_ID) for c in conflicts: print(" -", c) return 3 print("[%s] apply 完成，新建:" % BATCH_ID, created, "核对:", json.dumps(verify(cur, identity), ensure_ascii=False)) return 0 finally: conn.close() if __name__ == "__main__": sys.exit(main())