452 lines
15 KiB
Python
452 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""将「固体产品数据」导入生产库（合并同名同规格 + 多品牌/别名）。"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import uuid
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import pymysql
|
|
from dotenv import dotenv_values
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DATA_FILE = Path(__file__).resolve().parent / "固体产品数据"
|
|
ENV_FILE = ROOT / "backend" / ".env.production"
|
|
|
|
TENANT_NO = "shanxia2.0"
|
|
PRODUCT_NO_PREFIX = "SXWL"
|
|
CREATE_BY = "d3000000000000000000000000000001"
|
|
DEFAULT_UNIT = "dun"
|
|
DEFAULT_CATEGORY_NAME = "无机原料"
|
|
|
|
# 显式覆盖（用户确认）
|
|
CATEGORY_OVERRIDES = {
|
|
"甲苯": "有机原料",
|
|
}
|
|
|
|
|
|
def connect():
|
|
vals = dotenv_values(ENV_FILE)
|
|
u = urlparse(vals["DATABASE_URL"].replace("mysql+pymysql://", "mysql://"))
|
|
return pymysql.connect(
|
|
host=u.hostname,
|
|
port=u.port or 3306,
|
|
user=u.username,
|
|
password=u.password,
|
|
database=u.path.lstrip("/"),
|
|
charset="utf8mb4",
|
|
connect_timeout=15,
|
|
autocommit=False,
|
|
)
|
|
|
|
|
|
def norm_text(v: str) -> str:
|
|
v = (v or "").strip()
|
|
return "" if (not v or v == "/") else v
|
|
|
|
|
|
def fix_name(name: str) -> str:
|
|
name = name.strip()
|
|
if name.endswith(":"):
|
|
name = name.rstrip(":")
|
|
if name == "乙脂":
|
|
name = "乙酯"
|
|
return name
|
|
|
|
|
|
def fix_alias(name: str, alias: str) -> str:
|
|
alias = norm_text(alias)
|
|
if name == "聚丙烯酰胺" and alias == "PAC":
|
|
return "PAM"
|
|
return alias
|
|
|
|
|
|
def parse_groups(text: str) -> list[dict]:
|
|
"""按 名称+规格 合并。"""
|
|
groups: dict[tuple[str, str], dict] = {}
|
|
order: list[tuple[str, str]] = []
|
|
|
|
for i, line in enumerate(text.splitlines()):
|
|
if i == 0 or not line.strip():
|
|
continue
|
|
parts = line.split("\t")
|
|
while len(parts) < 6:
|
|
parts.append("")
|
|
name, brand, alias, model, tax, hazard = [p.strip() for p in parts[:6]]
|
|
if not name:
|
|
continue
|
|
name = fix_name(name)
|
|
brand = norm_text(brand)
|
|
alias = fix_alias(name, alias)
|
|
model = norm_text(model)
|
|
|
|
# 磷酸二氢钾：只留 98%
|
|
if name == "磷酸二氢钾" and model != "98%":
|
|
continue
|
|
|
|
key = (name, model)
|
|
if key not in groups:
|
|
groups[key] = {
|
|
"product_name": name,
|
|
"model": model or None,
|
|
"brands": [],
|
|
"aliases": [],
|
|
"tax_rate": Decimal(re.sub(r"[^\d.]", "", tax) or "13"),
|
|
"is_toxic": 0,
|
|
"is_explosive": 0,
|
|
}
|
|
order.append(key)
|
|
g = groups[key]
|
|
if brand and brand not in g["brands"]:
|
|
g["brands"].append(brand)
|
|
if alias and alias not in g["aliases"]:
|
|
g["aliases"].append(alias)
|
|
if "易制毒" in hazard:
|
|
g["is_toxic"] = 1
|
|
if "易制爆" in hazard:
|
|
g["is_explosive"] = 1
|
|
# 税率以首次为准；同组合应一致
|
|
if tax:
|
|
g["tax_rate"] = Decimal(re.sub(r"[^\d.]", "", tax) or "13")
|
|
|
|
return [groups[k] for k in order]
|
|
|
|
|
|
def load_live_categories(cur) -> dict[str, str]:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, category_name FROM crm_product_category
|
|
WHERE tenant_no=%s AND del_flag=0
|
|
""",
|
|
(TENANT_NO,),
|
|
)
|
|
return {name: cid for cid, name in cur.fetchall()}
|
|
|
|
|
|
def resolve_unit_and_category(cur, product_name: str, live_cats: dict[str, str]):
|
|
"""从下架同名产品解析单位、分类（多数决）。"""
|
|
# exact name
|
|
cur.execute(
|
|
"""
|
|
SELECT unit, COUNT(*) c
|
|
FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND product_name=%s
|
|
GROUP BY unit
|
|
ORDER BY c DESC, CASE WHEN unit='dun' THEN 0 ELSE 1 END
|
|
""",
|
|
(TENANT_NO, product_name),
|
|
)
|
|
unit_rows = cur.fetchall()
|
|
if not unit_rows:
|
|
# fuzzy helpers
|
|
if product_name == "碳酸氢钠":
|
|
cur.execute(
|
|
"""
|
|
SELECT unit, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND TRIM(product_name)=%s
|
|
GROUP BY unit ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, product_name),
|
|
)
|
|
unit_rows = cur.fetchall()
|
|
elif product_name == "聚丙烯酰胺":
|
|
cur.execute(
|
|
"""
|
|
SELECT unit, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off'
|
|
AND product_name LIKE %s
|
|
GROUP BY unit ORDER BY c DESC, CASE WHEN unit='dun' THEN 0 ELSE 1 END LIMIT 1
|
|
""",
|
|
(TENANT_NO, "聚丙烯酰胺%"),
|
|
)
|
|
unit_rows = cur.fetchall()
|
|
elif product_name == "乙二胺四乙酸二钠":
|
|
cur.execute(
|
|
"""
|
|
SELECT unit, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off'
|
|
AND product_name LIKE %s
|
|
GROUP BY unit ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, "乙二胺四乙酸二钠%"),
|
|
)
|
|
unit_rows = cur.fetchall()
|
|
elif product_name == "乙酯":
|
|
cur.execute(
|
|
"""
|
|
SELECT unit, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND product_name=%s
|
|
GROUP BY unit ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, "乙酸乙酯"),
|
|
)
|
|
unit_rows = cur.fetchall()
|
|
|
|
unit = unit_rows[0][0] if unit_rows and unit_rows[0][0] else DEFAULT_UNIT
|
|
|
|
# category by category_id majority
|
|
cur.execute(
|
|
"""
|
|
SELECT category_id, category_name, COUNT(*) c
|
|
FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND product_name=%s
|
|
AND category_id IS NOT NULL AND category_id != ''
|
|
GROUP BY category_id, category_name
|
|
ORDER BY c DESC
|
|
""",
|
|
(TENANT_NO, product_name),
|
|
)
|
|
cat_rows = cur.fetchall()
|
|
if not cat_rows:
|
|
if product_name == "碳酸氢钠":
|
|
cur.execute(
|
|
"""
|
|
SELECT category_id, category_name, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND TRIM(product_name)=%s
|
|
AND category_id IS NOT NULL AND category_id != ''
|
|
GROUP BY category_id, category_name ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, product_name),
|
|
)
|
|
cat_rows = cur.fetchall()
|
|
elif product_name == "聚丙烯酰胺":
|
|
cur.execute(
|
|
"""
|
|
SELECT category_id, category_name, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off'
|
|
AND product_name LIKE %s
|
|
AND category_id IS NOT NULL AND category_id != ''
|
|
GROUP BY category_id, category_name ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, "聚丙烯酰胺%"),
|
|
)
|
|
cat_rows = cur.fetchall()
|
|
elif product_name == "乙二胺四乙酸二钠":
|
|
cur.execute(
|
|
"""
|
|
SELECT category_id, category_name, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off'
|
|
AND product_name LIKE %s
|
|
AND category_id IS NOT NULL AND category_id != ''
|
|
GROUP BY category_id, category_name ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, "乙二胺四乙酸二钠%"),
|
|
)
|
|
cat_rows = cur.fetchall()
|
|
elif product_name == "乙酯":
|
|
cur.execute(
|
|
"""
|
|
SELECT category_id, category_name, COUNT(*) c FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND status='off' AND product_name=%s
|
|
AND category_id IS NOT NULL AND category_id != ''
|
|
GROUP BY category_id, category_name ORDER BY c DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, "乙酸乙酯"),
|
|
)
|
|
cat_rows = cur.fetchall()
|
|
|
|
if product_name in CATEGORY_OVERRIDES:
|
|
cname = CATEGORY_OVERRIDES[product_name]
|
|
cid = live_cats.get(cname)
|
|
if not cid:
|
|
raise RuntimeError(f"覆盖分类不存在: {cname}")
|
|
return unit, cid, cname
|
|
|
|
if cat_rows:
|
|
cid = cat_rows[0][0]
|
|
# 用当前分类表名称（水处理药剂 → 水处理溶剂）
|
|
for name, id_ in live_cats.items():
|
|
if id_ == cid:
|
|
return unit, cid, name
|
|
# ID 不在现行分类：回退名称映射
|
|
old_name = cat_rows[0][1] or ""
|
|
if old_name == "水处理药剂":
|
|
old_name = "水处理溶剂"
|
|
if old_name in live_cats:
|
|
return unit, live_cats[old_name], old_name
|
|
|
|
cname = DEFAULT_CATEGORY_NAME
|
|
return unit, live_cats[cname], cname
|
|
|
|
|
|
def main() -> None:
|
|
groups = parse_groups(DATA_FILE.read_text(encoding="utf-8"))
|
|
if not groups:
|
|
raise SystemExit("未解析到产品")
|
|
|
|
conn = connect()
|
|
cur = conn.cursor()
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
|
|
try:
|
|
live_cats = load_live_categories(cur)
|
|
if DEFAULT_CATEGORY_NAME not in live_cats:
|
|
raise RuntimeError("默认分类不存在")
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT MAX(CAST(SUBSTRING(product_no,5) AS UNSIGNED)) FROM crm_product
|
|
WHERE tenant_no=%s AND product_no REGEXP '^SXWL[0-9]+$'
|
|
""",
|
|
(TENANT_NO,),
|
|
)
|
|
max_seq = int(cur.fetchone()[0] or 0)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT barcode FROM crm_product
|
|
WHERE tenant_no=%s AND del_flag=0 AND barcode LIKE %s
|
|
ORDER BY barcode DESC LIMIT 1
|
|
""",
|
|
(TENANT_NO, f"{today}%"),
|
|
)
|
|
barcode_row = cur.fetchone()
|
|
if barcode_row and barcode_row[0] and str(barcode_row[0]).startswith(today):
|
|
try:
|
|
barcode_seq = int(str(barcode_row[0])[len(today) :]) + 1
|
|
except ValueError:
|
|
barcode_seq = 1
|
|
else:
|
|
barcode_seq = 1
|
|
|
|
imported = []
|
|
for idx, g in enumerate(groups):
|
|
unit, category_id, category_name = resolve_unit_and_category(
|
|
cur, g["product_name"], live_cats
|
|
)
|
|
max_seq += 1
|
|
product_no = f"{PRODUCT_NO_PREFIX}{max_seq:06d}"
|
|
barcode = f"{today}{barcode_seq:06d}"
|
|
barcode_seq += 1
|
|
pid = uuid.uuid4().hex.upper()
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO crm_product (
|
|
id, product_no, product_name, category_id, category_name,
|
|
model, unit, barcode, price, cost_price, stock, low_stock,
|
|
tax_rate, is_toxic, is_explosive, status, description, images,
|
|
sale_count, tenant_no, order_num, remark,
|
|
create_by, create_dept_id, create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
%s, %s, %s, %s, %s,
|
|
%s, %s, %s, NULL, NULL, 0, 10,
|
|
%s, %s, %s, 'on', NULL, NULL,
|
|
0, %s, %s, NULL,
|
|
%s, '', %s, %s, %s, 0
|
|
)
|
|
""",
|
|
(
|
|
pid,
|
|
product_no,
|
|
g["product_name"],
|
|
category_id,
|
|
category_name,
|
|
g["model"],
|
|
unit,
|
|
barcode,
|
|
g["tax_rate"],
|
|
g["is_toxic"],
|
|
g["is_explosive"],
|
|
TENANT_NO,
|
|
idx + 1,
|
|
CREATE_BY,
|
|
now,
|
|
CREATE_BY,
|
|
now,
|
|
),
|
|
)
|
|
|
|
for bidx, brand_name in enumerate(g["brands"]):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO crm_product_brand (
|
|
id, product_id, brand_name, status, tenant_no, order_num, remark,
|
|
create_by, create_dept_id, create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
%s, %s, %s, 'on', %s, %s, NULL,
|
|
%s, '', %s, %s, %s, 0
|
|
)
|
|
""",
|
|
(
|
|
uuid.uuid4().hex.upper(),
|
|
pid,
|
|
brand_name,
|
|
TENANT_NO,
|
|
bidx,
|
|
CREATE_BY,
|
|
now,
|
|
CREATE_BY,
|
|
now,
|
|
),
|
|
)
|
|
|
|
for aidx, alias_name in enumerate(g["aliases"]):
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO crm_product_alias (
|
|
id, product_id, alias_name, tenant_no, order_num, remark,
|
|
create_by, create_dept_id, create_time, update_by, update_time, del_flag
|
|
) VALUES (
|
|
%s, %s, %s, %s, %s, NULL,
|
|
%s, '', %s, %s, %s, 0
|
|
)
|
|
""",
|
|
(
|
|
uuid.uuid4().hex.upper(),
|
|
pid,
|
|
alias_name,
|
|
TENANT_NO,
|
|
aidx,
|
|
CREATE_BY,
|
|
now,
|
|
CREATE_BY,
|
|
now,
|
|
),
|
|
)
|
|
|
|
imported.append(
|
|
{
|
|
"product_no": product_no,
|
|
"product_name": g["product_name"],
|
|
"model": g["model"] or "",
|
|
"unit": unit,
|
|
"category_name": category_name,
|
|
"brands": ",".join(g["brands"]) or "-",
|
|
"aliases": ",".join(g["aliases"]) or "-",
|
|
"tax_rate": str(g["tax_rate"]),
|
|
"is_toxic": g["is_toxic"],
|
|
"is_explosive": g["is_explosive"],
|
|
}
|
|
)
|
|
|
|
conn.commit()
|
|
print(f"target=production/{TENANT_NO}")
|
|
print(f"imported={len(imported)}")
|
|
print("---")
|
|
print("编号\t名称\t规格\t单位\t分类\t品牌\t别名\t税率\t危化")
|
|
for item in imported:
|
|
hazard = []
|
|
if item["is_toxic"]:
|
|
hazard.append("易制毒")
|
|
if item["is_explosive"]:
|
|
hazard.append("易制爆")
|
|
print(
|
|
f"{item['product_no']}\t{item['product_name']}\t{item['model']}\t"
|
|
f"{item['unit']}\t{item['category_name']}\t{item['brands']}\t"
|
|
f"{item['aliases']}\t{item['tax_rate']}\t{','.join(hazard) or '-'}"
|
|
)
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|