chore(db): add optional script to split brand from product names
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
7aa995a9d6
commit
dd31fb53b6
1 file changed
+165
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
"""可选：从产品名称括号段拆出品牌。
|
||||
|
||||
示例：「片碱（君正）」→ product_name=片碱，crm_product_brand.brand_name=君正。
|
||||
|
||||
默认 dry-run：导出匹配清单 CSV，不改库。
|
||||
业务确认后加 --apply 才执行更新。
|
||||
|
||||
不修改历史报价/合同/计划行上的 product_name 快照。
|
||||
|
||||
用法:
|
||||
cd backend && ./venv/bin/python scripts/migrate_20260727_split_brand_from_product_name.py
|
||||
cd backend && ./venv/bin/python scripts/migrate_20260727_split_brand_from_product_name.py --csv /tmp/brands.csv
|
||||
cd backend && ./venv/bin/python scripts/migrate_20260727_split_brand_from_product_name.py --apply
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 允许直接 python scripts/... 运行
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from app.core.database import SessionLocal
|
||||
from app.core.utils import generate_uuid
|
||||
from app.models.crm_models import CrmProduct, CrmProductBrand
|
||||
|
||||
# 全角/半角括号均可：片碱（君正） / 片碱(君正)
|
||||
NAME_WITH_BRAND_RE = re.compile(
|
||||
r"^(?P<base>.+?)[（(](?P<brand>[^）)]+)[）)]\s*$"
|
||||
)
|
||||
|
||||
|
||||
def parse_name(name: str):
|
||||
text = (name or "").strip()
|
||||
m = NAME_WITH_BRAND_RE.match(text)
|
||||
if not m:
|
||||
return None
|
||||
base = m.group("base").strip()
|
||||
brand = m.group("brand").strip()
|
||||
if not base or not brand:
|
||||
return None
|
||||
return base, brand
|
||||
|
||||
|
||||
def collect_candidates(db):
|
||||
rows = (
|
||||
db.query(CrmProduct)
|
||||
.filter(CrmProduct.del_flag == 0)
|
||||
.order_by(CrmProduct.create_time.desc())
|
||||
.all()
|
||||
)
|
||||
out = []
|
||||
for row in rows:
|
||||
parsed = parse_name(row.product_name or "")
|
||||
if not parsed:
|
||||
continue
|
||||
base, brand = parsed
|
||||
out.append(
|
||||
{
|
||||
"id": row.id,
|
||||
"tenant_no": row.tenant_no,
|
||||
"product_no": row.product_no or "",
|
||||
"old_name": row.product_name,
|
||||
"new_name": base,
|
||||
"brand_name": brand,
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def write_csv(path: Path, rows: list[dict]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("w", encoding="utf-8-sig", newline="") as f:
|
||||
writer = csv.DictWriter(
|
||||
f,
|
||||
fieldnames=["id", "tenant_no", "product_no", "old_name", "new_name", "brand_name"],
|
||||
)
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
|
||||
|
||||
def apply_updates(db, rows: list[dict]) -> int:
|
||||
changed = 0
|
||||
for item in rows:
|
||||
product = (
|
||||
db.query(CrmProduct)
|
||||
.filter(
|
||||
CrmProduct.id == item["id"],
|
||||
CrmProduct.del_flag == 0,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not product:
|
||||
continue
|
||||
if (product.product_name or "").strip() != item["old_name"]:
|
||||
continue
|
||||
exists = (
|
||||
db.query(CrmProductBrand)
|
||||
.filter(
|
||||
CrmProductBrand.product_id == product.id,
|
||||
CrmProductBrand.tenant_no == product.tenant_no,
|
||||
CrmProductBrand.brand_name == item["brand_name"],
|
||||
CrmProductBrand.del_flag == 0,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not exists:
|
||||
db.add(
|
||||
CrmProductBrand(
|
||||
id=generate_uuid(),
|
||||
product_id=product.id,
|
||||
brand_name=item["brand_name"],
|
||||
status="on",
|
||||
tenant_no=product.tenant_no,
|
||||
order_num=0,
|
||||
create_by=product.update_by or product.create_by,
|
||||
create_dept_id=product.create_dept_id or "",
|
||||
update_by=product.update_by or product.create_by,
|
||||
)
|
||||
)
|
||||
product.product_name = item["new_name"]
|
||||
changed += 1
|
||||
db.commit()
|
||||
return changed
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="从产品名括号段拆品牌（需业务确认后 --apply）")
|
||||
parser.add_argument(
|
||||
"--csv",
|
||||
default="scripts/output/product_brand_split_candidates.csv",
|
||||
help="导出候选清单路径",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="确认后执行：改 product_name 并插入 crm_product_brand",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
rows = collect_candidates(db)
|
||||
csv_path = Path(args.csv)
|
||||
if not csv_path.is_absolute():
|
||||
csv_path = ROOT / csv_path
|
||||
write_csv(csv_path, rows)
|
||||
print(f"匹配 {len(rows)} 条，已导出: {csv_path}")
|
||||
if not args.apply:
|
||||
print("dry-run 完成。业务确认后加 --apply 再执行。")
|
||||
return
|
||||
changed = apply_updates(db, rows)
|
||||
print(f"已更新 {changed} 条产品名称，并补品牌子表（已存在同名品牌则跳过插入）。")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user