161 lines
5.4 KiB
Python
161 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
||
"""钉盘数据增量更新 - 下载新增CSV并导入数据库"""
|
||
import os, sys, json, subprocess, urllib.request, time
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
CSV_DIR = os.path.join(BASE_DIR, "tmp_csv")
|
||
os.makedirs(CSV_DIR, exist_ok=True)
|
||
|
||
def log(m): print(f"[{time.strftime('%H:%M:%S')}] {m}")
|
||
|
||
def dws(cmd):
|
||
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
|
||
try: return json.loads(r.stdout)
|
||
except: return {}
|
||
|
||
def download_file(node_id, filename):
|
||
"""下载单个文件"""
|
||
dest = os.path.join(CSV_DIR, filename)
|
||
if os.path.exists(dest):
|
||
log(f" 已存在，跳过: {filename}")
|
||
return True
|
||
log(f" 下载 {filename} ...")
|
||
# 用 --output 获取下载链接
|
||
tmp = f"/tmp/dld_{int(time.time())}_{filename}"
|
||
info = dws(f'dws doc download --node "{node_id}" --output "{tmp}" --format json')
|
||
if not info.get("success"):
|
||
log(f" ❌ 获取下载链接失败: {filename}")
|
||
return False
|
||
url = info.get("resourceUrl", "")
|
||
if not url:
|
||
log(f" ❌ 无下载链接: {filename}")
|
||
return False
|
||
try:
|
||
urllib.request.urlretrieve(url, dest)
|
||
size = os.path.getsize(dest)
|
||
log(f" ✅ {filename} ({size:,} bytes)")
|
||
# 清理临时文件
|
||
if os.path.exists(tmp): os.remove(tmp)
|
||
return True
|
||
except Exception as e:
|
||
log(f" ❌ 下载失败: {filename} - {e}")
|
||
return False
|
||
|
||
def get_all_nodes(folder_id):
|
||
"""获取文件夹下所有文件(自动翻页)"""
|
||
all_nodes = []
|
||
page_token = None
|
||
while True:
|
||
cmd = f'dws doc list --folder "{folder_id}" --format json'
|
||
if page_token:
|
||
cmd += f' --page-token "{page_token}"'
|
||
result = dws(cmd)
|
||
nodes = result.get('nodes', [])
|
||
all_nodes.extend(nodes)
|
||
if not result.get('hasMore'):
|
||
break
|
||
page_token = result.get('nextPageToken')
|
||
if not page_token:
|
||
break
|
||
return all_nodes
|
||
|
||
# ========== 1. 下载日出库数据（06-02 ~ 06-05）==========
|
||
log("=" * 60)
|
||
log("第一步：下载日出库 CSV 文件 (06-02 ~ 06-05)")
|
||
|
||
OUTBOUND_FOLDER = "3NwLYZXWynyBjRB4tZjwNxDvVkyEqBQm"
|
||
outbound_nodes = get_all_nodes(OUTBOUND_FOLDER)
|
||
log(f"日出库文件夹共 {len(outbound_nodes)} 个文件")
|
||
|
||
target_dates = ['20260602', '20260603', '20260604', '20260605']
|
||
downloaded_outbound = 0
|
||
|
||
for n in outbound_nodes:
|
||
name = n['name']
|
||
node_id = n.get('nodeId', '')
|
||
if any(name.startswith(d) for d in target_dates):
|
||
clean_name = name.strip().replace(' ', '')
|
||
if not clean_name.endswith('.csv'):
|
||
clean_name += '.csv'
|
||
if download_file(node_id, clean_name):
|
||
downloaded_outbound += 1
|
||
|
||
log(f"日出库下载完成: {downloaded_outbound} 个文件")
|
||
|
||
# ========== 2. 下载库存数据（06-08）==========
|
||
log("")
|
||
log("=" * 60)
|
||
log("第二步：下载库存 CSV 文件 (06-08)")
|
||
|
||
INVENTORY_FOLDER = "NkDwLng8ZLGrXmrnFxYq3vRqVKMEvZBY"
|
||
inventory_nodes = get_all_nodes(INVENTORY_FOLDER)
|
||
log(f"库存文件夹共 {len(inventory_nodes)} 个文件")
|
||
|
||
downloaded_inventory = 0
|
||
for n in inventory_nodes:
|
||
name = n['name']
|
||
node_id = n.get('nodeId', '')
|
||
if name.startswith('20260608') and name.endswith('.csv'):
|
||
if download_file(node_id, name):
|
||
downloaded_inventory += 1
|
||
|
||
log(f"库存下载完成: {downloaded_inventory} 个文件")
|
||
|
||
# ========== 3. 导入数据到MySQL ==========
|
||
log("")
|
||
log("=" * 60)
|
||
log("第三步：导入数据到数据库")
|
||
|
||
log("导入日出库数据...")
|
||
r = subprocess.run(
|
||
f"cd {BASE_DIR} && python3 reimport.py",
|
||
shell=True, capture_output=True, text=True, timeout=600
|
||
)
|
||
if r.returncode == 0:
|
||
log("✅ 日出库导入成功")
|
||
else:
|
||
log(f"⚠️ 日出库导入异常: {r.stderr[-300:]}")
|
||
for line in r.stdout.strip().split('\n')[-10:]:
|
||
log(f" {line}")
|
||
|
||
log("导入库存数据...")
|
||
r = subprocess.run(
|
||
f"cd {BASE_DIR} && python3 batch_import_inventory.py",
|
||
shell=True, capture_output=True, text=True, timeout=600
|
||
)
|
||
if r.returncode == 0:
|
||
log("✅ 库存导入成功")
|
||
else:
|
||
log(f"⚠️ 库存导入异常: {r.stderr[-300:]}")
|
||
for line in r.stdout.strip().split('\n')[-10:]:
|
||
log(f" {line}")
|
||
|
||
# ========== 4. 验证数据 ==========
|
||
log("")
|
||
log("=" * 60)
|
||
log("第四步：验证数据")
|
||
import pymysql
|
||
from config import DB_CONFIG
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
with conn.cursor() as cur:
|
||
cur.execute("SELECT MIN(日期) AS min_d, MAX(日期) AS max_d, COUNT(*) AS cnt FROM daily_outbound")
|
||
r = cur.fetchone()
|
||
log(f"日出库: {r[0]} ~ {r[1]}, 共 {r[2]} 条")
|
||
|
||
cur.execute("SELECT MIN(data_date) AS min_d, MAX(data_date) AS max_d, COUNT(*) AS cnt FROM inventory_detail")
|
||
r = cur.fetchone()
|
||
log(f"库存: {r[0]} ~ {r[1]}, 共 {r[2]} 条")
|
||
conn.close()
|
||
log("✅ 数据验证通过")
|
||
except Exception as e:
|
||
log(f"⚠️ 验证失败: {e}")
|
||
|
||
# ========== 5. 重启服务 ==========
|
||
log("")
|
||
log("=" * 60)
|
||
log("第五步：重启服务刷新缓存")
|
||
os.system("kill $(lsof -t -i:18888) 2>/dev/null; sleep 2")
|
||
os.system(f"cd {BASE_DIR} && nohup /home/gxd/.local/bin/gunicorn -w 4 -b 0.0.0.0:18888 --timeout 120 --max-requests 800 --max-requests-jitter 100 --access-logfile logs/access.log --error-logfile logs/error.log app:app > logs/startup.log 2>&1 &")
|
||
log("✅ 服务已重启，数据同步完成")
|