379 lines
13 KiB
Python
379 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import sys
|
||
from typing import Dict, List, Tuple
|
||
|
||
import pytds
|
||
|
||
BUSINESS_TABLES = [
|
||
"jst_goods_profit_kpi_items_afs",
|
||
"jst_sku",
|
||
"fys_shopInfo",
|
||
"jst_order_express_details",
|
||
"fys_SkuDataLibrary_SpuConfig",
|
||
"op_operManage",
|
||
"op_TargetManagement",
|
||
"fys_Ticketing_install",
|
||
"fys_Ticketing_install_item",
|
||
"fys_jstFreightCost",
|
||
"fys_jstFreightCostDetail",
|
||
"pc_kfChatLog",
|
||
]
|
||
|
||
ENV_FILE = "/root/.openclaw/credentials/company-sql.env"
|
||
|
||
|
||
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 conn_params() -> Dict[str, object]:
|
||
load_env_file(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": 300,
|
||
}
|
||
|
||
|
||
def get_conn():
|
||
params = conn_params()
|
||
if not params["password"]:
|
||
raise RuntimeError("COMPANY_DB_PASSWORD is empty")
|
||
return pytds.connect(**params)
|
||
|
||
|
||
def _fmt(value):
|
||
if isinstance(value, dt.datetime):
|
||
return value.isoformat(sep=" ")
|
||
if isinstance(value, (dt.date, dt.time)):
|
||
return value.isoformat()
|
||
return value
|
||
|
||
|
||
def print_rows(columns: List[str], rows: List[Tuple[object, ...]], truncated: bool = False) -> None:
|
||
print("COLUMNS:", ", ".join(columns))
|
||
if not rows:
|
||
print("ROWS: 0")
|
||
return
|
||
if truncated:
|
||
print(f"ROWS_SHOWN: {len(rows)} (truncated)")
|
||
else:
|
||
print(f"ROWS: {len(rows)}")
|
||
for row in rows:
|
||
obj = {columns[i]: _fmt(row[i]) for i in range(len(columns))}
|
||
print(json.dumps(obj, ensure_ascii=False, default=str))
|
||
|
||
|
||
def split_go_batches(sql_text: str) -> List[str]:
|
||
batches: List[str] = []
|
||
current: List[str] = []
|
||
for line in sql_text.splitlines():
|
||
if line.strip().upper() == "GO":
|
||
block = "\n".join(current).strip()
|
||
if block:
|
||
batches.append(block)
|
||
current = []
|
||
else:
|
||
current.append(line)
|
||
block = "\n".join(current).strip()
|
||
if block:
|
||
batches.append(block)
|
||
return batches
|
||
|
||
|
||
def fetch_with_limit(cursor, row_limit: int) -> Tuple[List[Tuple[object, ...]], bool]:
|
||
if row_limit <= 0:
|
||
return cursor.fetchall(), False
|
||
|
||
rows: List[Tuple[object, ...]] = []
|
||
truncated = False
|
||
while len(rows) <= row_limit:
|
||
batch = cursor.fetchmany(256)
|
||
if not batch:
|
||
break
|
||
rows.extend(batch)
|
||
if len(rows) > row_limit:
|
||
rows = rows[:row_limit]
|
||
truncated = True
|
||
break
|
||
return rows, truncated
|
||
|
||
|
||
def run_sql(sql_text: str, row_limit: int) -> int:
|
||
batches = split_go_batches(sql_text)
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
set_index = 0
|
||
for batch in batches:
|
||
cur.execute(batch)
|
||
while True:
|
||
if cur.description:
|
||
set_index += 1
|
||
cols = [c[0] for c in cur.description]
|
||
rows, truncated = fetch_with_limit(cur, row_limit)
|
||
print(f"RESULT_SET {set_index}")
|
||
print_rows(cols, rows, truncated=truncated)
|
||
if not cur.nextset():
|
||
break
|
||
cur.close()
|
||
return set_index
|
||
|
||
|
||
def cmd_health(_args: argparse.Namespace) -> int:
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT @@VERSION AS version, DB_NAME() AS db_name, GETDATE() AS server_time")
|
||
cols = [c[0] for c in cur.description]
|
||
row = cur.fetchone()
|
||
data = {cols[i]: _fmt(row[i]) for i in range(len(cols))}
|
||
print(json.dumps(data, ensure_ascii=False, default=str, indent=2))
|
||
cur.close()
|
||
return 0
|
||
|
||
|
||
def cmd_tables(_args: argparse.Namespace) -> int:
|
||
names = ", ".join([f"'{t}'" for t in BUSINESS_TABLES])
|
||
sql = f"""
|
||
SELECT c.TABLE_NAME, COUNT(*) AS column_count
|
||
FROM INFORMATION_SCHEMA.COLUMNS c
|
||
WHERE c.TABLE_SCHEMA='dbo'
|
||
AND c.TABLE_NAME IN ({names})
|
||
GROUP BY c.TABLE_NAME
|
||
ORDER BY c.TABLE_NAME
|
||
"""
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql)
|
||
rows = cur.fetchall()
|
||
cols = [c[0] for c in cur.description]
|
||
print_rows(cols, rows)
|
||
cur.close()
|
||
return 0
|
||
|
||
|
||
def cmd_columns(args: argparse.Namespace) -> int:
|
||
table = args.table
|
||
sql = f"""
|
||
SELECT
|
||
COLUMN_NAME,
|
||
DATA_TYPE,
|
||
CHARACTER_MAXIMUM_LENGTH,
|
||
NUMERIC_PRECISION,
|
||
NUMERIC_SCALE,
|
||
IS_NULLABLE,
|
||
ORDINAL_POSITION
|
||
FROM INFORMATION_SCHEMA.COLUMNS
|
||
WHERE TABLE_SCHEMA='dbo' AND TABLE_NAME='{table}'
|
||
ORDER BY ORDINAL_POSITION
|
||
"""
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql)
|
||
rows = cur.fetchall()
|
||
cols = [c[0] for c in cur.description]
|
||
print_rows(cols, rows)
|
||
cur.close()
|
||
return 0
|
||
|
||
|
||
def _collect_schema() -> Dict[str, List[Tuple[object, ...]]]:
|
||
names = ", ".join([f"'{t}'" for t in BUSINESS_TABLES])
|
||
sql = f"""
|
||
SELECT
|
||
TABLE_NAME,
|
||
COLUMN_NAME,
|
||
DATA_TYPE,
|
||
CHARACTER_MAXIMUM_LENGTH,
|
||
NUMERIC_PRECISION,
|
||
NUMERIC_SCALE,
|
||
IS_NULLABLE,
|
||
ORDINAL_POSITION
|
||
FROM INFORMATION_SCHEMA.COLUMNS
|
||
WHERE TABLE_SCHEMA='dbo'
|
||
AND TABLE_NAME IN ({names})
|
||
ORDER BY TABLE_NAME, ORDINAL_POSITION
|
||
"""
|
||
with get_conn() as conn:
|
||
cur = conn.cursor()
|
||
cur.execute(sql)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
by_table: Dict[str, List[Tuple[object, ...]]] = {t: [] for t in BUSINESS_TABLES}
|
||
for row in rows:
|
||
by_table[str(row[0])].append(row)
|
||
return by_table
|
||
|
||
|
||
def cmd_catalog(args: argparse.Namespace) -> int:
|
||
by_table = _collect_schema()
|
||
now = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
out = pathlib.Path(args.out)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
lines: List[str] = []
|
||
lines.append("# 公司销售/售后数据库字段与维度目录")
|
||
lines.append("")
|
||
lines.append(f"> 自动生成时间: {now}")
|
||
lines.append("> 数据源: SQLServer / JSDataMiddlePlatform / dbo")
|
||
lines.append("> 适用范围: 公司销售数据、售后数据、店铺运营、物流运费、安装工单、客服问题")
|
||
lines.append("")
|
||
lines.append("## 一、强制路由规则（给AI）")
|
||
lines.append("")
|
||
lines.append("- 凡是涉及 `销售/业绩/售后/退款/费比/店铺业绩` 的需求，必须走数据库查询，禁止走旧业绩网页。")
|
||
lines.append("- 先查字段目录再写 SQL：`/root/.openclaw/workspace/outputs/company-sales-db-catalog.md`")
|
||
lines.append("- 优先复用周报模板：`/root/.openclaw/workspace/sqlserver/last_week_afs_report.sql`")
|
||
lines.append("- 输出结论必须标注：`时间范围 + 使用表 + 关键字段 + 过滤条件`。")
|
||
lines.append("")
|
||
lines.append("## 二、可查询业务维度（语义层）")
|
||
lines.append("")
|
||
dimension_rows = [
|
||
("订单维度", "订单号、子订单号、内部订单ID、订单状态、售后状态", "jst_order_express_details, jst_goods_profit_kpi_items_afs"),
|
||
("商品维度", "SKU、SPU、货号、商品名称、规格、类目、品牌", "jst_sku, op_TargetManagement, op_operManage, fys_SkuDataLibrary_SpuConfig"),
|
||
("店铺维度", "店铺名称、平台、店铺编码、负责人、渠道类型", "fys_shopInfo, jst_order_express_details, jst_goods_profit_kpi_items_afs"),
|
||
("销售金额维度", "商品金额、付款金额、实发金额、成本、毛利相关费用", "jst_goods_profit_kpi_items_afs, jst_sku"),
|
||
("售后维度", "售后类型、售后原因、退款/退货金额、售后标签", "jst_goods_profit_kpi_items_afs, pc_kfChatLog"),
|
||
("物流维度", "快递公司、快递单号、物流状态、运费、重量、收货地区", "jst_order_express_details, fys_jstFreightCost, fys_jstFreightCostDetail"),
|
||
("安装工单维度", "工单状态、安装费用、结算金额、服务商", "fys_Ticketing_install, fys_Ticketing_install_item"),
|
||
("人员维度", "运营、设计师、采购、美工、店长、客服", "op_TargetManagement, op_operManage, fys_shopInfo, fys_SkuDataLibrary_SpuConfig"),
|
||
("时间维度", "创建时间、支付时间、发货时间、签收时间、订单完成时间", "全表通用时间字段（createdAt/createTime/pay_time/send_time 等）"),
|
||
("客服问题维度", "聊天主次原因、问题内容、订单关联", "pc_kfChatLog"),
|
||
]
|
||
lines.append("| 维度 | 可查询内容 | 主要来源表 |")
|
||
lines.append("|---|---|---|")
|
||
for row in dimension_rows:
|
||
lines.append(f"| {row[0]} | {row[1]} | {row[2]} |")
|
||
lines.append("")
|
||
lines.append("## 三、核心关联键（Join Key）")
|
||
lines.append("")
|
||
lines.append("- `order_id`：`so_id / undefined2 / orderNo / OrderNum / Sourceno`")
|
||
lines.append("- `inner_order_id`：`inner_order_id / undefined1 / o_id`")
|
||
lines.append("- `sku_id`：`sku_id / undefined10 / barCode`")
|
||
lines.append("- `spu`：`spu / other_1 / undefined96`")
|
||
lines.append("- `shop_name`：`shop_name / shopName / undefined8 / shopname`")
|
||
lines.append("")
|
||
lines.append("## 四、全表字段清单（真实表头）")
|
||
lines.append("")
|
||
|
||
total_cols = 0
|
||
for table in BUSINESS_TABLES:
|
||
rows = by_table.get(table, [])
|
||
total_cols += len(rows)
|
||
lines.append(f"### {table}（{len(rows)} 列）")
|
||
lines.append("")
|
||
lines.append("| 序号 | 字段名 | 类型 | 长度/精度 | 可空 |")
|
||
lines.append("|---:|---|---|---|---|")
|
||
for r in rows:
|
||
_, col, dtype, clen, prec, scale, nullable, ord_pos = r
|
||
if clen is not None:
|
||
length = str(clen)
|
||
elif prec is not None:
|
||
if scale is not None:
|
||
length = f"{prec},{scale}"
|
||
else:
|
||
length = str(prec)
|
||
else:
|
||
length = "-"
|
||
lines.append(f"| {ord_pos} | {col} | {dtype} | {length} | {nullable} |")
|
||
lines.append("")
|
||
|
||
lines.append("## 五、统计")
|
||
lines.append("")
|
||
lines.append(f"- 业务表数量：{len(BUSINESS_TABLES)}")
|
||
lines.append(f"- 字段总数：{total_cols}")
|
||
lines.append("- 建议：高频分析优先在订单售后主表 `jst_goods_profit_kpi_items_afs` 先聚合，再按需关联其他表，避免超时。")
|
||
|
||
out.write_text("\n".join(lines), encoding="utf-8")
|
||
print(str(out))
|
||
return 0
|
||
|
||
|
||
def cmd_query(args: argparse.Namespace) -> int:
|
||
sql_text = None
|
||
|
||
if args.file:
|
||
sql_text = pathlib.Path(args.file).read_text(encoding="utf-8")
|
||
elif args.sql:
|
||
sql_text = args.sql
|
||
elif args.sql_positional:
|
||
sql_text = " ".join(args.sql_positional).strip()
|
||
|
||
if not sql_text:
|
||
raise SystemExit("query requires --sql / --file / or inline SQL text")
|
||
|
||
set_count = run_sql(sql_text, args.limit)
|
||
print(f"TOTAL_RESULT_SETS: {set_count}")
|
||
return 0
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(prog="company-db", description="Company SQLServer helper")
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
sub.add_parser("health")
|
||
sub.add_parser("tables")
|
||
|
||
p_cols = sub.add_parser("columns")
|
||
p_cols.add_argument("table")
|
||
|
||
p_query = sub.add_parser("query")
|
||
p_query.add_argument("--sql", help="SQL text")
|
||
p_query.add_argument("--file", help="SQL file path")
|
||
p_query.add_argument("--limit", type=int, default=200, help="row limit via SET ROWCOUNT")
|
||
p_query.add_argument("sql_positional", nargs="*", help="inline SQL text, e.g. company-db query SELECT 1")
|
||
|
||
p_run = sub.add_parser("run-file")
|
||
p_run.add_argument("file", help="SQL file path")
|
||
p_run.add_argument("--limit", type=int, default=0, help="max rows shown per result set (0 means no limit)")
|
||
|
||
p_cat = sub.add_parser("catalog")
|
||
p_cat.add_argument("--out", default="/root/.openclaw/workspace/outputs/company-sales-db-catalog.md")
|
||
|
||
return p
|
||
|
||
|
||
def main() -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args()
|
||
|
||
if args.cmd == "health":
|
||
return cmd_health(args)
|
||
if args.cmd == "tables":
|
||
return cmd_tables(args)
|
||
if args.cmd == "columns":
|
||
return cmd_columns(args)
|
||
if args.cmd == "query":
|
||
return cmd_query(args)
|
||
if args.cmd == "run-file":
|
||
args.sql = None
|
||
args.file = args.file
|
||
return cmd_query(args)
|
||
if args.cmd == "catalog":
|
||
return cmd_catalog(args)
|
||
|
||
parser.print_help()
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|