shanxia2.0/产品数据刷新/import_liquid_products.py

#!/usr/bin/env python3
"""将「液体产品数据」导入生产库 crm_product（租户 shanxia2.0）。"""
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"
UNIT = "dun"
PRODUCT_NO_PREFIX = "SXWL"
CREATE_BY = "d3000000000000000000000000000001"


def parse_rows(text: str) -> list[dict]:
    rows: list[dict] = []
    for i, line in enumerate(text.splitlines()):
        line = line.strip("\n")
        if i == 0 and line.startswith("产品名称"):
            continue
        if not line.strip():
            continue
        parts = line.split("\t")
        while len(parts) < 5:
            parts.append("")
        name, category, model, tax, hazard = [p.strip() for p in parts[:5]]
        if not name:
            continue
        tax_num = Decimal(re.sub(r"[^\d.]", "", tax) or "13")
        is_toxic = 1 if "易制毒" in hazard else 0
        is_explosive = 1 if "易制爆" in hazard else 0
        rows.append(
            {
                "product_name": name,
                "category_name": category,
                "model": model,
                "tax_rate": tax_num,
                "is_toxic": is_toxic,
                "is_explosive": is_explosive,
            }
        )
    return rows


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 main() -> None:
    rows = parse_rows(DATA_FILE.read_text(encoding="utf-8"))
    if not rows:
        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:
        # 分类：无机原料
        cur.execute(
            """
            SELECT id, category_name FROM crm_product_category
            WHERE tenant_no=%s AND del_flag=0 AND category_name=%s
            LIMIT 1
            """,
            (TENANT_NO, "无机原料"),
        )
        cat = cur.fetchone()
        if not cat:
            raise RuntimeError("未找到分类「无机原料」")
        category_id, category_name = cat

        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, row in enumerate(rows):
            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,
                    row["product_name"],
                    category_id,
                    category_name,
                    row["model"] or None,
                    UNIT,
                    barcode,
                    row["tax_rate"],
                    row["is_toxic"],
                    row["is_explosive"],
                    TENANT_NO,
                    idx + 1,
                    CREATE_BY,
                    now,
                    CREATE_BY,
                    now,
                ),
            )
            imported.append(
                {
                    "product_no": product_no,
                    "product_name": row["product_name"],
                    "model": row["model"],
                    "unit": UNIT,
                    "tax_rate": str(row["tax_rate"]),
                    "is_toxic": row["is_toxic"],
                    "is_explosive": row["is_explosive"],
                    "category_name": category_name,
                    "id": pid,
                }
            )

        conn.commit()
        print(f"target=production/{TENANT_NO}")
        print(f"category={category_name}({category_id})")
        print(f"imported={len(imported)}")
        print("---")
        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['tax_rate']}\t{','.join(hazard) or '-'}"
            )
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


if __name__ == "__main__":
    main()