314 lines
11 KiB
Python
314 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
订单汇总 A–X 全量镜像刷新
|
||
触发：Step2（Cursor Step1 完成后 @小溪）
|
||
归属：小溪 (xiaoxi)
|
||
|
||
进表条件：K=是 · O>0 · 非全额退(P空或P<O) · L≥C
|
||
全额退 → 整行不进订单表（销售行清 K/O/P/Q）
|
||
镜像 A–V 原样 + W 渠道归属 + X=1
|
||
|
||
分工约定见 docs/bot-step2-schedule-and-orders.md
|
||
"""
|
||
import json, time, re, sys, requests, psycopg2
|
||
from datetime import datetime
|
||
from feishu_sheet_utils import FeishuSheetWriter
|
||
|
||
# ── 配置 ──
|
||
APP_ID = "cli_a929ae22e0b8dcc8"
|
||
APP_SECRET = "OtFjMy7p3qE3VvLbMdcWidwgHOnGD4FJ"
|
||
SPREADSHEET_TOKEN = "NoZqsFi47hIOHEt9j8WcfRtbnug"
|
||
SALES_SHEETS = {"f975f0": "吴迪", "qJF4I": "小龙", "qJF4J": "成都"}
|
||
DIRECT_SHEET = "1sosYE" # 直购表（小红书店铺+stream-xhs，不依赖手机号匹配）
|
||
SUMMARY_SHEET = "2smjwA"
|
||
|
||
def _get_pg_password():
|
||
import os
|
||
secrets_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "secrets.env")
|
||
with open(secrets_path) as f:
|
||
for line in f:
|
||
if line.startswith("PG_ONLINE_PASSWORD="):
|
||
return line.strip().split("=", 1)[1].strip('"').strip("'")
|
||
raise RuntimeError("PG_ONLINE_PASSWORD not found in secrets.env")
|
||
|
||
PG_CONFIG = {
|
||
"host": "bj-postgres-16pob4sg.sql.tencentcdb.com", "port": 28591,
|
||
"user": "ai_member", "password": _get_pg_password(), "database": "vala_bi",
|
||
}
|
||
|
||
GOODS_MAP = {
|
||
57: "瓦拉英语level1·单季", 60: "瓦拉英语level1", 63: "瓦拉英语level1·单季",
|
||
31: "瓦拉英语年包", 32: "瓦拉英语单季度包", 33: "瓦拉英语level2", 54: "瓦拉英语季度包",
|
||
61: "瓦拉英语level1+2",
|
||
}
|
||
|
||
# 达人昵称关键词
|
||
DAREN_NICKNAMES = ["晚柠", "学霸", "念妈", "神奇瓜妈", "三人行", "老王"]
|
||
|
||
|
||
def get_token():
|
||
r = requests.post("https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
|
||
json={"app_id": APP_ID, "app_secret": APP_SECRET}, timeout=15)
|
||
return r.json()["tenant_access_token"]
|
||
|
||
|
||
def read_sheet(token, sheet_id, range_str):
|
||
url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values/{sheet_id}!{range_str}?valueRenderOption=ToString"
|
||
r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=30)
|
||
data = r.json()
|
||
if data.get("code") != 0:
|
||
print(f"Error reading {sheet_id}: {data}")
|
||
return []
|
||
return data["data"]["valueRange"]["values"]
|
||
|
||
|
||
def put_values(token, sheet_id, range_str, values, retries=3):
|
||
url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{SPREADSHEET_TOKEN}/values"
|
||
body = {"valueRange": {"range": f"{sheet_id}!{range_str}", "values": values}}
|
||
for attempt in range(retries):
|
||
r = requests.put(url, headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
}, json=body, timeout=30)
|
||
result = r.json()
|
||
if result.get("code") == 0:
|
||
return True
|
||
print(f" Retry {attempt+1} for {range_str}: {result.get('msg','')}")
|
||
time.sleep(1)
|
||
print(f" FAILED {range_str}")
|
||
return False
|
||
|
||
|
||
def parse_date(s):
|
||
"""Parse date string to (year, month, day) tuple."""
|
||
s = str(s).strip()
|
||
if not s:
|
||
return None
|
||
# YYYY-MM-DD
|
||
m = re.match(r'(\d{4})-(\d{1,2})-(\d{1,2})', s)
|
||
if m:
|
||
return (int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||
# M月D日
|
||
m = re.match(r'(\d{1,2})月(\d{1,2})日', s)
|
||
if m:
|
||
return (2026, int(m.group(1)), int(m.group(2)))
|
||
# YYYY/M/D
|
||
m = re.match(r'(\d{4})/(\d{1,2})/(\d{1,2})', s)
|
||
if m:
|
||
return (int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
||
return None
|
||
|
||
|
||
def date_le(a, b):
|
||
"""Return True if date a <= date b."""
|
||
if a is None or b is None:
|
||
return False
|
||
return a <= b
|
||
|
||
|
||
def classify_w_channel(m_channel, sales_name=""):
|
||
"""
|
||
W 渠道归属，基于 M 成交渠道 + 销售昵称。
|
||
达人：newmedia-daren/jingxuan/stream-daren 或昵称含达人关键词
|
||
销转：sales-adp-* / stream-wxxd-* / 含「微信小店」
|
||
端内：app-active/app-sales/partner-actives 或 M=端内
|
||
直购：dianpu-* / stream-xhs-* / 抖音/直购/小红书
|
||
其他：其余
|
||
"""
|
||
m = str(m_channel).strip() if m_channel else ""
|
||
ml = m.lower()
|
||
|
||
# 达人
|
||
if any(kw in ml for kw in ["newmedia-daren", "jingxuan", "stream-daren"]):
|
||
return "达人"
|
||
if any(nick in str(sales_name) for nick in DAREN_NICKNAMES):
|
||
return "达人"
|
||
|
||
# 销转
|
||
if any(kw in ml for kw in ["sales-adp", "stream-wxxd"]):
|
||
return "销转"
|
||
if "微信小店" in m:
|
||
return "销转"
|
||
|
||
# 端内
|
||
if any(kw in ml for kw in ["app-active", "app-sales", "partner-active"]):
|
||
return "端内"
|
||
if m == "端内":
|
||
return "端内"
|
||
|
||
# 直购
|
||
if any(kw in ml for kw in ["dianpu-", "stream-xhs"]):
|
||
return "直购"
|
||
if any(kw in m for kw in ["抖音", "直购", "小红书"]):
|
||
return "直购"
|
||
|
||
return "其他"
|
||
|
||
|
||
def phone_match(sheet_phone, db_tel):
|
||
"""Match sheet phone number against DB tel (masked like 138****4503)."""
|
||
if not sheet_phone or not db_tel:
|
||
return False
|
||
sheet_phone = str(sheet_phone).strip()
|
||
db_tel = str(db_tel).strip()
|
||
if sheet_phone == db_tel:
|
||
return True
|
||
if "****" in db_tel:
|
||
parts = db_tel.split("****")
|
||
if len(parts) == 2:
|
||
prefix, suffix = parts
|
||
if sheet_phone.startswith(prefix) and sheet_phone.endswith(suffix):
|
||
return True
|
||
return False
|
||
|
||
|
||
def main():
|
||
print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] 订单汇总全量刷新 启动")
|
||
token = get_token()
|
||
|
||
# ── Step 1: 读取销售三表 ──
|
||
all_rows = []
|
||
for sid, name in SALES_SHEETS.items():
|
||
print(f"Reading {name}...")
|
||
vals = read_sheet(token, sid, "A3:V10000")
|
||
filtered = []
|
||
for i, row in enumerate(vals):
|
||
while len(row) < 22:
|
||
row.append("")
|
||
b = str(row[1]).strip() if row[1] else ""
|
||
e = str(row[4]).strip() if row[4] else ""
|
||
h = str(row[7]).strip() if row[7] else ""
|
||
if b or e or h:
|
||
filtered.append({
|
||
"sid": sid, "name": name, "row": i + 3,
|
||
"raw": row[:22], # A-V
|
||
})
|
||
print(f" {len(filtered)} non-empty rows")
|
||
all_rows.extend(filtered)
|
||
|
||
print(f"Total rows: {len(all_rows)}")
|
||
|
||
# ── Step 1.5: 读取直购表 ──
|
||
print(f"Reading 直购...")
|
||
direct_vals = read_sheet(token, DIRECT_SHEET, "A2:V10000")
|
||
direct_rows = []
|
||
for i, row in enumerate(direct_vals):
|
||
while len(row) < 22:
|
||
row.append("")
|
||
b = str(row[1]).strip() if row[1] else ""
|
||
h = str(row[7]).strip() if row[7] else ""
|
||
if b or h:
|
||
direct_rows.append({
|
||
"sid": DIRECT_SHEET, "name": "直购", "row": i + 2,
|
||
"raw": row[:22],
|
||
})
|
||
print(f" {len(direct_rows)} non-empty rows")
|
||
all_rows.extend(direct_rows)
|
||
print(f"Total rows (with 直购): {len(all_rows)}")
|
||
|
||
# ── Step 2: 筛选进订单汇总的行 ──
|
||
# 条件：K=是 · O>0 · 非全额退(P空或P<O) · L≥C
|
||
order_rows = []
|
||
for r in all_rows:
|
||
raw = r["raw"]
|
||
k = str(raw[10]).strip() if len(raw) > 10 and raw[10] else ""
|
||
if k != "是":
|
||
continue
|
||
|
||
# O > 0
|
||
try:
|
||
o_val = float(raw[14]) if len(raw) > 14 and raw[14] not in (None, "") else 0
|
||
except (ValueError, TypeError):
|
||
o_val = 0
|
||
if o_val <= 0:
|
||
continue
|
||
|
||
# 非全额退: P空或P<O
|
||
try:
|
||
p_val = float(raw[15]) if len(raw) > 15 and raw[15] not in (None, "") else 0
|
||
except (ValueError, TypeError):
|
||
p_val = 0
|
||
if p_val > 0 and p_val >= o_val:
|
||
# 全额退 → 不进订单表
|
||
continue
|
||
|
||
# L ≥ C (C为空时跳过此检查，如直购用户无进线日期)
|
||
c_str = str(raw[2]).strip() if len(raw) > 2 and raw[2] else ""
|
||
l_str = str(raw[11]).strip() if len(raw) > 11 and raw[11] else ""
|
||
c_date = parse_date(c_str)
|
||
l_date = parse_date(l_str)
|
||
if c_date is not None and not date_le(c_date, l_date):
|
||
continue
|
||
|
||
# 通过所有条件
|
||
order_rows.append(r)
|
||
|
||
print(f"Order rows after filter: {len(order_rows)}")
|
||
|
||
# ── Step 2.5: 去重（同一人可能在三表+直购中出现多次）──
|
||
# 按 (A销售归属, B微信昵称, O下单金额, P退款金额, L下单日期) 去重
|
||
seen = set()
|
||
deduped = []
|
||
for r in order_rows:
|
||
raw = r["raw"]
|
||
a = str(raw[0]).strip() if raw[0] else ""
|
||
b = str(raw[1]).strip() if len(raw) > 1 and raw[1] else ""
|
||
o = str(raw[14]).strip() if len(raw) > 14 and raw[14] else ""
|
||
p = str(raw[15]).strip() if len(raw) > 15 and raw[15] else ""
|
||
l = str(raw[11]).strip() if len(raw) > 11 and raw[11] else ""
|
||
key = (a, b, o, p, l)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
deduped.append(r)
|
||
dup_count = len(order_rows) - len(deduped)
|
||
if dup_count > 0:
|
||
print(f" Removed {dup_count} duplicate rows")
|
||
order_rows = deduped
|
||
|
||
# ── Step 3: 按 L 下单日降序 ──
|
||
order_rows.sort(key=lambda r: str(r["raw"][11]) if len(r["raw"]) > 11 and r["raw"][11] else "", reverse=True)
|
||
|
||
# ── Step 4: 构建 A–X 行 ──
|
||
summary_rows = []
|
||
for r in order_rows:
|
||
raw = r["raw"]
|
||
# A–V 原样镜像
|
||
new_row = list(raw[:22])
|
||
|
||
# W: 渠道归属（基于 M 成交渠道）
|
||
m_channel = str(raw[12]).strip() if len(raw) > 12 and raw[12] else ""
|
||
sales_name = str(raw[0]).strip() if len(raw) > 0 and raw[0] else ""
|
||
w = classify_w_channel(m_channel, sales_name)
|
||
new_row.append(w)
|
||
|
||
# X: 有效成单 = 1
|
||
new_row.append(1)
|
||
|
||
summary_rows.append(new_row)
|
||
|
||
print(f"Summary rows: {len(summary_rows)}")
|
||
|
||
# ── Step 5: 写入订单汇总（使用安全写入工具，自动遵守 5000 格上限）──
|
||
print("Writing to 订单汇总...")
|
||
writer = FeishuSheetWriter(SPREADSHEET_TOKEN, token)
|
||
|
||
# 先清空旧数据区（26 列，自动计算批大小 ≤ 4400 格/批）
|
||
writer.clear(SUMMARY_SHEET, start_row=3, end_row=2000, cols=26)
|
||
time.sleep(0.5)
|
||
|
||
# 写入新数据（24 列 A-X，自动分批）
|
||
total = len(summary_rows)
|
||
writer.write(SUMMARY_SHEET, start_row=3, rows=summary_rows, cols=24)
|
||
|
||
# ── Step 6: 清除多余旧行 ──
|
||
existing = read_sheet(token, SUMMARY_SHEET, "A3:A4000")
|
||
old_count = len([r for r in existing if r and any(c for c in r if c)])
|
||
if old_count > total:
|
||
writer.clear(SUMMARY_SHEET, start_row=3 + total, end_row=3 + old_count - 1, cols=24)
|
||
|
||
print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] ✅ 订单汇总刷新完成")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|