You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
866 lines
45 KiB
866 lines
45 KiB
#!/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())
|
|
|