860 lines
29 KiB
Python
860 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import re
|
||
import sys
|
||
from typing import Dict, List, Sequence, Tuple
|
||
|
||
import pytds
|
||
import pymysql
|
||
|
||
SQL_ENV_FILE = "/root/.openclaw/credentials/company-sql.env"
|
||
MYSQL_ENV_FILE = "/root/.openclaw/credentials/install-media-mysql.env"
|
||
DEFAULT_MAPPING_FILE = "/root/.openclaw/workspace/outputs/install-media-mysql-mapping.json"
|
||
|
||
|
||
def load_env_file(path: str) -> None:
|
||
p = pathlib.Path(path)
|
||
if not p.exists():
|
||
return
|
||
for raw in p.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, val = line.split("=", 1)
|
||
key = key.strip()
|
||
val = val.strip().strip('"').strip("'")
|
||
if key and key not in os.environ:
|
||
os.environ[key] = val
|
||
|
||
|
||
def sql_conn_params() -> Dict[str, object]:
|
||
load_env_file(SQL_ENV_FILE)
|
||
return {
|
||
"server": os.environ.get("COMPANY_DB_HOST", "218.93.191.16"),
|
||
"port": int(os.environ.get("COMPANY_DB_PORT", "1433")),
|
||
"database": os.environ.get("COMPANY_DB_NAME", "JSDataMiddlePlatform"),
|
||
"user": os.environ.get("COMPANY_DB_USER", "nanting"),
|
||
"password": os.environ.get("COMPANY_DB_PASSWORD", ""),
|
||
"login_timeout": 15,
|
||
"timeout": 120,
|
||
}
|
||
|
||
|
||
def mysql_conn_params() -> Dict[str, object]:
|
||
load_env_file(MYSQL_ENV_FILE)
|
||
return {
|
||
"host": os.environ.get("INSTALL_MEDIA_MYSQL_HOST", "218.93.191.20"),
|
||
"port": int(os.environ.get("INSTALL_MEDIA_MYSQL_PORT", "3306")),
|
||
"user": os.environ.get("INSTALL_MEDIA_MYSQL_USER", "read_only_jiansun"),
|
||
"password": os.environ.get("INSTALL_MEDIA_MYSQL_PASSWORD", ""),
|
||
"charset": "utf8mb4",
|
||
"connect_timeout": int(os.environ.get("INSTALL_MEDIA_MYSQL_CONNECT_TIMEOUT", "8")),
|
||
"read_timeout": int(os.environ.get("INSTALL_MEDIA_MYSQL_READ_TIMEOUT", "15")),
|
||
"write_timeout": int(os.environ.get("INSTALL_MEDIA_MYSQL_WRITE_TIMEOUT", "15")),
|
||
"autocommit": True,
|
||
}
|
||
|
||
|
||
def sql_escape(value: str) -> str:
|
||
return value.replace("'", "''")
|
||
|
||
|
||
def qi(identifier: str) -> str:
|
||
return "`" + identifier.replace("`", "``") + "`"
|
||
|
||
|
||
def fmt_value(value):
|
||
if isinstance(value, (dt.datetime, dt.date, dt.time)):
|
||
return value.isoformat(sep=" ")
|
||
return value
|
||
|
||
|
||
def print_rows(columns: Sequence[str], rows: Sequence[Dict[str, object]]) -> None:
|
||
print("COLUMNS:", ", ".join(columns))
|
||
if not rows:
|
||
print("ROWS: 0")
|
||
return
|
||
print(f"ROWS: {len(rows)}")
|
||
for row in rows:
|
||
obj = {c: fmt_value(row.get(c)) for c in columns}
|
||
print(json.dumps(obj, ensure_ascii=False, default=str))
|
||
|
||
|
||
def query_order_product_mappings(order_id: str, limit: int) -> List[Dict[str, object]]:
|
||
if not order_id:
|
||
return []
|
||
|
||
params = sql_conn_params()
|
||
if not params["password"]:
|
||
raise RuntimeError("COMPANY_DB_PASSWORD is empty")
|
||
|
||
oid = sql_escape(order_id.strip())
|
||
lim = max(1, min(limit, 2000))
|
||
|
||
sql = f"""
|
||
USE [JSDataMiddlePlatform];
|
||
SET NOCOUNT ON;
|
||
|
||
DECLARE @order_id nvarchar(120) = N'{oid}';
|
||
|
||
WITH order_map AS (
|
||
SELECT
|
||
NULLIF(a.undefined2, '') AS order_id,
|
||
NULLIF(a.undefined10, '') AS sku_id,
|
||
COALESCE(NULLIF(s.other_7, ''), NULLIF(a.undefined96, '')) AS product_no,
|
||
COALESCE(NULLIF(s.name, ''), NULLIF(a.undefined36, '')) AS sku_name,
|
||
COALESCE(NULLIF(a.undefined8, ''), NULLIF(e.shop_name, '')) AS shop_name,
|
||
COALESCE(
|
||
TRY_CONVERT(datetime2, NULLIF(a.undefined6, ''), 120),
|
||
TRY_CONVERT(datetime2, REPLACE(NULLIF(a.undefined6, ''), '/', '-'), 120),
|
||
TRY_CONVERT(datetime2, NULLIF(e.send_time, ''), 120),
|
||
TRY_CONVERT(datetime2, a.createdAt)
|
||
) AS send_time,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY NULLIF(a.undefined2, ''), COALESCE(NULLIF(s.other_7, ''), NULLIF(a.undefined96, ''), NULLIF(a.undefined10, ''))
|
||
ORDER BY COALESCE(
|
||
TRY_CONVERT(datetime2, NULLIF(a.undefined6, ''), 120),
|
||
TRY_CONVERT(datetime2, REPLACE(NULLIF(a.undefined6, ''), '/', '-'), 120),
|
||
TRY_CONVERT(datetime2, NULLIF(e.send_time, ''), 120),
|
||
TRY_CONVERT(datetime2, a.createdAt)
|
||
) DESC,
|
||
a.createdAt DESC
|
||
) AS rn
|
||
FROM dbo.jst_goods_profit_kpi_items_afs AS a WITH (NOLOCK)
|
||
LEFT JOIN dbo.jst_sku AS s WITH (NOLOCK)
|
||
ON s.sku_id = NULLIF(a.undefined10, '')
|
||
LEFT JOIN dbo.jst_order_express_details AS e WITH (NOLOCK)
|
||
ON NULLIF(e.so_id, '') = NULLIF(a.undefined2, '')
|
||
WHERE (
|
||
NULLIF(a.undefined2, '') = @order_id
|
||
OR NULLIF(a.undefined3, '') = @order_id
|
||
OR NULLIF(a.undefined3, '') LIKE @order_id + '%'
|
||
)
|
||
AND ISNULL(NULLIF(a.undefined62, ''), '') NOT LIKE '%配件%'
|
||
AND ISNULL(NULLIF(a.undefined62, ''), '') NOT LIKE '%特殊单%'
|
||
)
|
||
SELECT TOP ({lim})
|
||
order_id,
|
||
sku_id,
|
||
product_no,
|
||
sku_name,
|
||
shop_name,
|
||
send_time
|
||
FROM order_map
|
||
WHERE rn = 1
|
||
AND ISNULL(NULLIF(order_id, ''), '') <> ''
|
||
AND ISNULL(NULLIF(product_no, ''), '') <> ''
|
||
ORDER BY send_time DESC, order_id ASC;
|
||
"""
|
||
|
||
with pytds.connect(**params) as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql)
|
||
cols = [c[0] for c in cur.description]
|
||
data = [dict(zip(cols, row)) for row in cur.fetchall()]
|
||
cur.close()
|
||
return data
|
||
|
||
|
||
def _normalize_text(s: str) -> str:
|
||
return (s or "").strip().lower()
|
||
|
||
|
||
PRODUCT_COL_KEYS = [
|
||
"productno",
|
||
"product_no",
|
||
"goodsno",
|
||
"goods_no",
|
||
"itemno",
|
||
"item_no",
|
||
"stylecode",
|
||
"style_code",
|
||
"spu",
|
||
"sku",
|
||
"barcode",
|
||
"procode",
|
||
"pro_code",
|
||
"货号",
|
||
"商品编码",
|
||
"产品编码",
|
||
]
|
||
|
||
DRAWING_COL_KEYS = [
|
||
"instruction",
|
||
"drawing",
|
||
"manual",
|
||
"install",
|
||
"tuzhi",
|
||
"图纸",
|
||
"安装图",
|
||
"说明书",
|
||
]
|
||
|
||
VIDEO_COL_KEYS = [
|
||
"video",
|
||
"mp4",
|
||
"movie",
|
||
"教程",
|
||
"视频",
|
||
"安装视频",
|
||
]
|
||
|
||
PATH_COL_KEYS = [
|
||
"path",
|
||
"url",
|
||
"file",
|
||
"img",
|
||
"image",
|
||
"地址",
|
||
"链接",
|
||
"文件",
|
||
"路径",
|
||
]
|
||
|
||
SYSTEM_SCHEMAS = {"information_schema", "mysql", "performance_schema", "sys"}
|
||
|
||
|
||
def _contains_any(text: str, keys: Sequence[str]) -> bool:
|
||
low = _normalize_text(text)
|
||
return any(k in low for k in keys)
|
||
|
||
|
||
def discover_table_specs(
|
||
conn,
|
||
explicit_schema: str,
|
||
explicit_table: str,
|
||
max_tables: int,
|
||
) -> List[Dict[str, object]]:
|
||
where = ["TABLE_SCHEMA NOT IN ('information_schema','mysql','performance_schema','sys')"]
|
||
params: List[object] = []
|
||
if explicit_schema:
|
||
where.append("TABLE_SCHEMA = %s")
|
||
params.append(explicit_schema)
|
||
if explicit_table:
|
||
where.append("TABLE_NAME = %s")
|
||
params.append(explicit_table)
|
||
|
||
sql = f"""
|
||
SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, COALESCE(COLUMN_COMMENT, '')
|
||
FROM information_schema.COLUMNS
|
||
WHERE {' AND '.join(where)}
|
||
ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION
|
||
"""
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
rows = cur.fetchall()
|
||
|
||
grouped: Dict[Tuple[str, str], List[Tuple[str, str, str]]] = {}
|
||
for schema, table, col, dtype, comment in rows:
|
||
key = (schema, table)
|
||
grouped.setdefault(key, []).append((str(col), str(dtype), str(comment or "")))
|
||
|
||
specs: List[Dict[str, object]] = []
|
||
for (schema, table), cols in grouped.items():
|
||
product_cols: List[str] = []
|
||
drawing_cols: List[str] = []
|
||
video_cols: List[str] = []
|
||
generic_media_cols: List[str] = []
|
||
|
||
for col_name, _dtype, col_comment in cols:
|
||
text = f"{col_name} {col_comment}"
|
||
has_path = _contains_any(text, PATH_COL_KEYS)
|
||
has_drawing = _contains_any(text, DRAWING_COL_KEYS)
|
||
has_video = _contains_any(text, VIDEO_COL_KEYS)
|
||
has_product = _contains_any(text, PRODUCT_COL_KEYS)
|
||
|
||
if has_product:
|
||
product_cols.append(col_name)
|
||
|
||
if has_drawing and (has_path or has_drawing):
|
||
drawing_cols.append(col_name)
|
||
elif has_video and (has_path or has_video):
|
||
video_cols.append(col_name)
|
||
elif has_path and (has_drawing or has_video):
|
||
generic_media_cols.append(col_name)
|
||
elif has_path and _contains_any(text, ["install", "instruction", "图纸", "视频"]):
|
||
generic_media_cols.append(col_name)
|
||
|
||
media_cols = sorted(set(drawing_cols + video_cols + generic_media_cols))
|
||
if not product_cols or not media_cols:
|
||
continue
|
||
|
||
specs.append(
|
||
{
|
||
"schema": schema,
|
||
"table": table,
|
||
"product_cols": sorted(set(product_cols)),
|
||
"drawing_cols": sorted(set(drawing_cols)),
|
||
"video_cols": sorted(set(video_cols)),
|
||
"media_cols": media_cols,
|
||
}
|
||
)
|
||
|
||
specs.sort(key=lambda x: (x["schema"], x["table"]))
|
||
if max_tables > 0:
|
||
specs = specs[:max_tables]
|
||
return specs
|
||
|
||
|
||
def load_mapping_specs(mapping_file: str) -> List[Dict[str, object]]:
|
||
p = pathlib.Path(mapping_file)
|
||
if not p.exists():
|
||
return []
|
||
try:
|
||
obj = json.loads(p.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return []
|
||
|
||
rows = obj.get("tables") if isinstance(obj, dict) else None
|
||
if not isinstance(rows, list):
|
||
return []
|
||
|
||
specs: List[Dict[str, object]] = []
|
||
for row in rows:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
schema = str(row.get("schema", "")).strip()
|
||
table = str(row.get("table", "")).strip()
|
||
product_cols = [str(x).strip() for x in row.get("product_cols", []) if str(x).strip()]
|
||
drawing_cols = [str(x).strip() for x in row.get("drawing_cols", []) if str(x).strip()]
|
||
video_cols = [str(x).strip() for x in row.get("video_cols", []) if str(x).strip()]
|
||
media_cols = [str(x).strip() for x in row.get("media_cols", []) if str(x).strip()]
|
||
if not schema or not table or not product_cols:
|
||
continue
|
||
merged_media = sorted(set(media_cols + drawing_cols + video_cols))
|
||
if not merged_media:
|
||
continue
|
||
specs.append(
|
||
{
|
||
"schema": schema,
|
||
"table": table,
|
||
"product_cols": sorted(set(product_cols)),
|
||
"drawing_cols": sorted(set(drawing_cols)),
|
||
"video_cols": sorted(set(video_cols)),
|
||
"media_cols": merged_media,
|
||
}
|
||
)
|
||
return specs
|
||
|
||
|
||
def classify_media_type(col_name: str, value: str) -> str:
|
||
text = _normalize_text(f"{col_name} {value}")
|
||
if _contains_any(text, VIDEO_COL_KEYS) or re.search(r"\.(mp4|avi|mov|mkv|wmv)(\?|$)", text):
|
||
return "video"
|
||
if _contains_any(text, DRAWING_COL_KEYS) or re.search(r"\.(png|jpg|jpeg|webp|gif|bmp)(\?|$)", text):
|
||
return "drawing"
|
||
return "unknown"
|
||
|
||
|
||
def normalize_media_url(raw_path: str, local_prefix: str, cdn_prefix: str) -> str:
|
||
s = (raw_path or "").strip()
|
||
if not s:
|
||
return ""
|
||
|
||
if s.startswith("http://") or s.startswith("https://"):
|
||
return s
|
||
|
||
s_norm = s.replace("\\", "/")
|
||
local_norm = (local_prefix or "D:/JianSunTuZhiFile").replace("\\", "/").rstrip("/")
|
||
cdn_norm = (cdn_prefix or "https://img.jiansuntuzhi.cn/").rstrip("/") + "/"
|
||
|
||
if s_norm.lower().startswith(local_norm.lower() + "/"):
|
||
rest = s_norm[len(local_norm) + 1 :]
|
||
return cdn_norm + rest
|
||
if s_norm.lower() == local_norm.lower():
|
||
return cdn_norm.rstrip("/")
|
||
|
||
marker = "jiansuntuzhifile/"
|
||
idx = s_norm.lower().find(marker)
|
||
if idx >= 0:
|
||
rest = s_norm[idx + len(marker) :]
|
||
return cdn_norm + rest
|
||
|
||
if re.match(r"^[a-zA-Z]:/", s_norm):
|
||
return s_norm
|
||
|
||
if s_norm.startswith("/"):
|
||
return cdn_norm + s_norm.lstrip("/")
|
||
|
||
return cdn_norm + s_norm
|
||
|
||
|
||
def _query_table_rows(conn, sql: str, params: Sequence[object]) -> List[Dict[str, object]]:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
cols = [d[0] for d in cur.description]
|
||
data = [dict(zip(cols, row)) for row in cur.fetchall()]
|
||
return data
|
||
|
||
|
||
def query_jiansun_special_records(
|
||
conn,
|
||
schema: str,
|
||
product_no: str,
|
||
row_limit: int,
|
||
local_prefix: str,
|
||
cdn_prefix: str,
|
||
) -> Tuple[List[Dict[str, object]], List[Dict[str, object]]]:
|
||
"""
|
||
Deterministic path for jiansunblueprint:
|
||
product_no -> sheet1 -> missiontask -> missiontask_instructions/missiontask_video
|
||
"""
|
||
table_rows: List[Dict[str, object]] = []
|
||
detail_rows: List[Dict[str, object]] = []
|
||
|
||
if not schema:
|
||
return table_rows, detail_rows
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute(f"SHOW TABLES FROM {qi(schema)}")
|
||
table_names = {str(r[0]).lower() for r in cur.fetchall()}
|
||
|
||
required = {"sheet1", "missiontask", "missiontask_instructions", "missiontask_video"}
|
||
if not required.issubset(table_names):
|
||
return table_rows, detail_rows
|
||
|
||
lim = max(1, min(row_limit, 2000))
|
||
pno = product_no.strip()
|
||
|
||
# 1) sheet1: map product_no -> barcode/pictureId/sku model
|
||
sheet_sql = (
|
||
f"SELECT {qi('sku(货号)')} AS product_no, {qi('工厂条形码')} AS barcode, "
|
||
f"{qi('图纸ID')} AS picture_id, {qi('SKU型号')} AS sku_model "
|
||
f"FROM {qi(schema)}.{qi('sheet1')} "
|
||
f"WHERE {qi('sku(货号)')} = %s OR {qi('sku(货号)')} LIKE %s "
|
||
f"LIMIT %s"
|
||
)
|
||
sheet_rows = _query_table_rows(conn, sheet_sql, (pno, f"{pno}%", lim))
|
||
table_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"source_schema": schema,
|
||
"source_table": "sheet1",
|
||
"matched_rows": len(sheet_rows),
|
||
"matched_media_items": 0,
|
||
}
|
||
)
|
||
|
||
barcodes = sorted({str(r.get("barcode", "")).strip() for r in sheet_rows if str(r.get("barcode", "")).strip()})
|
||
picture_ids = sorted({str(r.get("picture_id", "")).strip() for r in sheet_rows if str(r.get("picture_id", "")).strip()})
|
||
sku_models = sorted({str(r.get("sku_model", "")).strip() for r in sheet_rows if str(r.get("sku_model", "")).strip()})
|
||
|
||
# 2) missiontask: find task ids
|
||
match_parts: List[str] = []
|
||
params: List[object] = []
|
||
|
||
match_parts.append(f"{qi('skucodes')} LIKE %s")
|
||
params.append(f"%{pno}%")
|
||
|
||
for b in barcodes[:60]:
|
||
match_parts.append(f"{qi('skucodes')} LIKE %s")
|
||
params.append(f"%{b}%")
|
||
for pid in picture_ids[:60]:
|
||
match_parts.append(f"{qi('pictureId')} = %s")
|
||
params.append(pid)
|
||
for sm in sku_models[:40]:
|
||
match_parts.append(f"{qi('skuModel')} = %s")
|
||
params.append(sm)
|
||
|
||
where_clause = (
|
||
f"({qi('isDelete')} = '0' OR {qi('isDelete')} = 0 OR {qi('isDelete')} IS NULL)"
|
||
f" AND ({' OR '.join(match_parts)})"
|
||
)
|
||
|
||
mission_sql = (
|
||
f"SELECT {qi('id')} AS task_id, {qi('taskId')} AS task_no, {qi('SPU')} AS spu, "
|
||
f"{qi('picture')} AS picture_path, {qi('pictureId')} AS picture_id, {qi('skucodes')} AS skucodes, "
|
||
f"{qi('withInstructions')} AS with_instructions, {qi('withVideo')} AS with_video, "
|
||
f"{qi('createdTime')} AS created_time "
|
||
f"FROM {qi(schema)}.{qi('missiontask')} "
|
||
f"WHERE {where_clause} "
|
||
f"ORDER BY {qi('createdTime')} DESC "
|
||
f"LIMIT %s"
|
||
)
|
||
params.append(max(20, min(lim * 3, 500)))
|
||
mission_rows = _query_table_rows(conn, mission_sql, params)
|
||
table_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"source_schema": schema,
|
||
"source_table": "missiontask",
|
||
"matched_rows": len(mission_rows),
|
||
"matched_media_items": 0,
|
||
}
|
||
)
|
||
|
||
task_ids = [str(r.get("task_id", "")).strip() for r in mission_rows if str(r.get("task_id", "")).strip()]
|
||
if not task_ids:
|
||
return table_rows, detail_rows
|
||
|
||
# 3) instructions
|
||
placeholders = ",".join(["%s"] * len(task_ids))
|
||
ins_sql = (
|
||
f"SELECT {qi('id')} AS id, {qi('hid')} AS task_id, {qi('url')} AS raw_path, {qi('orderNum')} AS order_num "
|
||
f"FROM {qi(schema)}.{qi('missiontask_instructions')} "
|
||
f"WHERE {qi('hid')} IN ({placeholders}) AND {qi('isDelete')} = 0 "
|
||
f"ORDER BY {qi('hid')} ASC, {qi('orderNum')} ASC"
|
||
)
|
||
ins_rows = _query_table_rows(conn, ins_sql, task_ids)
|
||
for r in ins_rows:
|
||
raw_path = str(r.get("raw_path", "")).strip()
|
||
if not raw_path:
|
||
continue
|
||
detail_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"media_type": "drawing",
|
||
"media_url": normalize_media_url(raw_path, local_prefix, cdn_prefix),
|
||
"raw_path": raw_path,
|
||
"source_schema": schema,
|
||
"source_table": "missiontask_instructions",
|
||
"source_column": "url",
|
||
}
|
||
)
|
||
|
||
# 4) video
|
||
video_sql = (
|
||
f"SELECT {qi('id')} AS id, {qi('hid')} AS task_id, {qi('url')} AS raw_path, "
|
||
f"{qi('videoId')} AS video_id, {qi('orderNum')} AS order_num "
|
||
f"FROM {qi(schema)}.{qi('missiontask_video')} "
|
||
f"WHERE {qi('hid')} IN ({placeholders}) AND {qi('isDelete')} = 0 "
|
||
f"ORDER BY {qi('hid')} ASC, {qi('orderNum')} ASC"
|
||
)
|
||
video_rows = _query_table_rows(conn, video_sql, task_ids)
|
||
for r in video_rows:
|
||
raw_path = str(r.get("raw_path", "")).strip()
|
||
video_id = str(r.get("video_id", "")).strip()
|
||
media_url = ""
|
||
source_col = "url"
|
||
if raw_path:
|
||
media_url = normalize_media_url(raw_path, local_prefix, cdn_prefix)
|
||
elif video_id:
|
||
# youku id format: XNjUyMTkzNjkxMg==
|
||
media_url = f"https://v.youku.com/v_show/id_{video_id}.html"
|
||
raw_path = video_id
|
||
source_col = "videoId"
|
||
else:
|
||
continue
|
||
|
||
detail_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"media_type": "video",
|
||
"media_url": media_url,
|
||
"raw_path": raw_path,
|
||
"source_schema": schema,
|
||
"source_table": "missiontask_video",
|
||
"source_column": source_col,
|
||
}
|
||
)
|
||
|
||
table_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"source_schema": schema,
|
||
"source_table": "missiontask_instructions",
|
||
"matched_rows": len(ins_rows),
|
||
"matched_media_items": sum(1 for x in detail_rows if x.get("source_table") == "missiontask_instructions"),
|
||
}
|
||
)
|
||
table_rows.append(
|
||
{
|
||
"product_no": pno,
|
||
"source_schema": schema,
|
||
"source_table": "missiontask_video",
|
||
"matched_rows": len(video_rows),
|
||
"matched_media_items": sum(1 for x in detail_rows if x.get("source_table") == "missiontask_video"),
|
||
}
|
||
)
|
||
return table_rows, detail_rows
|
||
|
||
|
||
def query_media_records(
|
||
conn,
|
||
specs: Sequence[Dict[str, object]],
|
||
product_no: str,
|
||
row_limit: int,
|
||
local_prefix: str,
|
||
cdn_prefix: str,
|
||
) -> Tuple[List[Dict[str, object]], List[Dict[str, object]]]:
|
||
detail_rows: List[Dict[str, object]] = []
|
||
table_rows: List[Dict[str, object]] = []
|
||
|
||
for spec in specs:
|
||
schema = str(spec["schema"])
|
||
table = str(spec["table"])
|
||
product_cols = list(spec["product_cols"])
|
||
media_cols = list(spec["media_cols"])
|
||
drawing_cols = set(spec.get("drawing_cols", []))
|
||
video_cols = set(spec.get("video_cols", []))
|
||
|
||
select_cols = sorted(set(product_cols + media_cols))
|
||
where_parts: List[str] = []
|
||
params: List[object] = []
|
||
for col in product_cols:
|
||
qcol = qi(col)
|
||
where_parts.append(f"({qcol} = %s OR {qcol} LIKE %s)")
|
||
params.extend([product_no, f"{product_no}%"])
|
||
|
||
if not where_parts:
|
||
continue
|
||
|
||
sql = (
|
||
f"SELECT {', '.join(qi(c) for c in select_cols)} "
|
||
f"FROM {qi(schema)}.{qi(table)} "
|
||
f"WHERE {' OR '.join(where_parts)} "
|
||
f"LIMIT %s"
|
||
)
|
||
params.append(max(1, min(row_limit, 2000)))
|
||
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
data = cur.fetchall()
|
||
|
||
if not data:
|
||
continue
|
||
|
||
col_index = {name: i for i, name in enumerate(select_cols)}
|
||
matched_media = 0
|
||
|
||
for row in data:
|
||
product_val = ""
|
||
for pcol in product_cols:
|
||
value = row[col_index[pcol]]
|
||
if value is not None and str(value).strip():
|
||
product_val = str(value).strip()
|
||
break
|
||
|
||
for mcol in media_cols:
|
||
value = row[col_index[mcol]]
|
||
if value is None:
|
||
continue
|
||
raw_path = str(value).strip()
|
||
if not raw_path:
|
||
continue
|
||
|
||
media_type = classify_media_type(mcol, raw_path)
|
||
if mcol in drawing_cols:
|
||
media_type = "drawing"
|
||
elif mcol in video_cols:
|
||
media_type = "video"
|
||
|
||
normalized = normalize_media_url(raw_path, local_prefix, cdn_prefix)
|
||
detail_rows.append(
|
||
{
|
||
"product_no": product_val or product_no,
|
||
"media_type": media_type,
|
||
"media_url": normalized,
|
||
"raw_path": raw_path,
|
||
"source_schema": schema,
|
||
"source_table": table,
|
||
"source_column": mcol,
|
||
}
|
||
)
|
||
matched_media += 1
|
||
|
||
table_rows.append(
|
||
{
|
||
"product_no": product_no,
|
||
"source_schema": schema,
|
||
"source_table": table,
|
||
"matched_rows": len(data),
|
||
"matched_media_items": matched_media,
|
||
}
|
||
)
|
||
|
||
return table_rows, detail_rows
|
||
|
||
|
||
def dedupe_media_rows(rows: Sequence[Dict[str, object]]) -> List[Dict[str, object]]:
|
||
seen = set()
|
||
out: List[Dict[str, object]] = []
|
||
for row in rows:
|
||
key = (
|
||
row.get("product_no", ""),
|
||
row.get("media_type", ""),
|
||
row.get("media_url", ""),
|
||
row.get("raw_path", ""),
|
||
row.get("source_schema", ""),
|
||
row.get("source_table", ""),
|
||
row.get("source_column", ""),
|
||
)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(dict(row))
|
||
return out
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(
|
||
description="Lookup install drawing/video links by order-id -> product-no -> MySQL media DB"
|
||
)
|
||
parser.add_argument("--order-id", default="", help="商品订单号")
|
||
parser.add_argument("--product-no", default="", help="产品货号（可跳过订单映射直接查）")
|
||
parser.add_argument("--sku-id", default="", help="SKU编码（等价于 --product-no，直接在安装库中匹配）")
|
||
parser.add_argument("--limit", type=int, default=200, help="每个来源表最多查询行数")
|
||
parser.add_argument("--max-tables", type=int, default=120, help="自动发现时最多探测表数")
|
||
parser.add_argument("--mysql-schema", default="", help="指定MySQL schema，减少扫描范围")
|
||
parser.add_argument("--mysql-table", default="", help="指定MySQL table，减少扫描范围")
|
||
parser.add_argument("--mapping-file", default=DEFAULT_MAPPING_FILE, help="固定映射配置JSON")
|
||
parser.add_argument("--local-prefix", default=os.environ.get("INSTALL_MEDIA_LOCAL_PREFIX", "D:\\JianSunTuZhiFile"), help="本地路径前缀")
|
||
parser.add_argument("--cdn-prefix", default=os.environ.get("INSTALL_MEDIA_CDN_PREFIX", "https://img.jiansuntuzhi.cn/"), help="图纸/视频CDN前缀")
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
order_id = args.order_id.strip()
|
||
direct_product = args.product_no.strip() or args.sku_id.strip()
|
||
|
||
if not order_id and not direct_product:
|
||
print("STATUS: failed")
|
||
print("ERROR: missing order-id or product-no")
|
||
return 2
|
||
|
||
warnings: List[str] = []
|
||
mapping_rows: List[Dict[str, object]] = []
|
||
|
||
if order_id:
|
||
try:
|
||
mapping_rows.extend(query_order_product_mappings(order_id, limit=300))
|
||
except Exception as exc:
|
||
warnings.append(f"SQLSERVER_ORDER_MAPPING_FAILED: {exc}")
|
||
|
||
if direct_product:
|
||
mapping_rows.append(
|
||
{
|
||
"order_id": order_id or "(direct)",
|
||
"sku_id": "",
|
||
"product_no": direct_product,
|
||
"sku_name": "",
|
||
"shop_name": "",
|
||
"send_time": None,
|
||
}
|
||
)
|
||
|
||
# dedupe mapping by (order_id, product_no, sku_id)
|
||
dedup = {}
|
||
for row in mapping_rows:
|
||
k = (str(row.get("order_id", "")).strip(), str(row.get("product_no", "")).strip(), str(row.get("sku_id", "")).strip())
|
||
if not k[1]:
|
||
continue
|
||
dedup[k] = row
|
||
mapping_rows = list(dedup.values())
|
||
|
||
product_nos = sorted({str(r.get("product_no", "")).strip() for r in mapping_rows if str(r.get("product_no", "")).strip()})
|
||
|
||
table_summary_rows: List[Dict[str, object]] = []
|
||
media_rows: List[Dict[str, object]] = []
|
||
|
||
if product_nos:
|
||
mysql_params = mysql_conn_params()
|
||
if not mysql_params["password"]:
|
||
warnings.append("MYSQL_PASSWORD_EMPTY: please set INSTALL_MEDIA_MYSQL_PASSWORD")
|
||
else:
|
||
try:
|
||
with pymysql.connect(**mysql_params) as conn:
|
||
schema_hint = args.mysql_schema.strip() or "jiansunblueprint"
|
||
source_mode = "special_jiansun"
|
||
discovered_count = 0
|
||
|
||
for pno in product_nos:
|
||
ts, ms = query_jiansun_special_records(
|
||
conn=conn,
|
||
schema=schema_hint,
|
||
product_no=pno,
|
||
row_limit=max(1, args.limit),
|
||
local_prefix=args.local_prefix,
|
||
cdn_prefix=args.cdn_prefix,
|
||
)
|
||
table_summary_rows.extend(ts)
|
||
media_rows.extend(ms)
|
||
|
||
if not ms:
|
||
specs = load_mapping_specs(args.mapping_file)
|
||
if specs:
|
||
source_mode = "mapping_file_fallback"
|
||
else:
|
||
specs = discover_table_specs(
|
||
conn,
|
||
args.mysql_schema.strip(),
|
||
args.mysql_table.strip(),
|
||
max(1, args.max_tables),
|
||
)
|
||
source_mode = "auto_discovery_fallback"
|
||
discovered_count = len(specs)
|
||
ts2, ms2 = query_media_records(
|
||
conn=conn,
|
||
specs=specs,
|
||
product_no=pno,
|
||
row_limit=max(1, args.limit),
|
||
local_prefix=args.local_prefix,
|
||
cdn_prefix=args.cdn_prefix,
|
||
)
|
||
table_summary_rows.extend(ts2)
|
||
media_rows.extend(ms2)
|
||
|
||
print(f"DISCOVERY_MODE: {source_mode}")
|
||
print(f"DISCOVERED_TABLES: {discovered_count}")
|
||
except Exception as exc:
|
||
warnings.append(f"MYSQL_LOOKUP_FAILED: {exc}")
|
||
else:
|
||
warnings.append("NO_PRODUCT_NO_FROM_ORDER: no product_no mapped from order-id")
|
||
|
||
media_rows = dedupe_media_rows(media_rows)
|
||
|
||
status = "ok"
|
||
if not media_rows and warnings:
|
||
status = "partial"
|
||
if not media_rows and not product_nos:
|
||
status = "failed"
|
||
|
||
print(f"STATUS: {status}")
|
||
print(f"ORDER_ID: {order_id or '(none)'}")
|
||
print(f"PRODUCT_NO_COUNT: {len(product_nos)}")
|
||
|
||
print("RESULT_SET 1")
|
||
print("NAME: order_to_product_mapping")
|
||
print_rows(
|
||
["order_id", "sku_id", "product_no", "sku_name", "shop_name", "send_time"],
|
||
mapping_rows,
|
||
)
|
||
|
||
print("RESULT_SET 2")
|
||
print("NAME: mysql_table_match_summary")
|
||
print_rows(
|
||
["product_no", "source_schema", "source_table", "matched_rows", "matched_media_items"],
|
||
table_summary_rows,
|
||
)
|
||
|
||
print("RESULT_SET 3")
|
||
print("NAME: install_media_links")
|
||
print_rows(
|
||
["product_no", "media_type", "media_url", "raw_path", "source_schema", "source_table", "source_column"],
|
||
media_rows,
|
||
)
|
||
|
||
if warnings:
|
||
print("WARNINGS:")
|
||
for w in warnings:
|
||
print(f"- {w}")
|
||
|
||
if not media_rows:
|
||
print("NEXT_STEP:")
|
||
if warnings:
|
||
print("- verify MySQL network reachability and fixed table mapping file")
|
||
else:
|
||
print("- matched sku/product but no install drawing/video rows in current tables")
|
||
|
||
return 0 if status in {"ok", "partial"} else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|