diff --git a/scripts/ding_daily_sync.py b/scripts/ding_daily_sync.py new file mode 100644 index 0000000..25637c2 --- /dev/null +++ b/scripts/ding_daily_sync.py @@ -0,0 +1,470 @@ +#!/usr/bin/env python3 +"""Transactional DingDrive -> cloud -> LAN replication. No app/network monkeypatch imports.""" +import argparse +import csv +import fcntl +import gzip +import hashlib +import importlib.util +import io +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +import time +from datetime import datetime, date, timedelta +from decimal import Decimal +from zoneinfo import ZoneInfo + +import pymysql + +ROOT = Path(__file__).resolve().parents[1] +STATE = ROOT / 'var/ding-sync' +SSH = ['ssh', '-i', '/home/dieya/.ssh/id_ed25519_dycg_restore', '-o', 'BatchMode=yes', + '-o', 'ConnectTimeout=15', 'ubuntu@82.156.135.91'] +REMOTE = '/opt/dy-dashboard/var/ding-sync' +DWS = '/home/ubuntu/.npm-global/lib/node_modules/dingtalk-workspace-cli/vendor/dws' +SPACE = '28814565790' +FOLDERS = {'inventory': 'NkDwLng8ZLGrXmrnFxYq3vRqVKMEvZBY', + 'outbound': '3NwLYZXWynyBjRB4tZjwNxDvVkyEqBQm'} +SOURCES = {'wdt', 'jdbz', 'jdgj', 'jdjwt', 'tmgj', 'dymuto', 'dyqsimelr', 'jdsmj', 'jdmrt'} +PATTERNS = {'inventory': re.compile(r'^(20\d{6})-wdt-kcmx-([\w]+)\.csv$'), + 'outbound': re.compile(r'^(20\d{6})-([a-z0-9]+)-cksj\.csv$')} +COLS = {'inventory': ['data_date', 'source', 'warehouse_code', 'goods_code', 'product_code', + 'product_name', 'brand', 'cost_price', 'available_qty', 'production_in_transit', + 'purchase_in_transit', 'category1', 'category2', 'category3', 'created_time'], + 'outbound': ['货品编号', '货品名称', '品牌', '仓库', '店铺', '一级类目', '二级类目', + '三级类目', '货品数量', '货品成本', '日期', '数据来源', '数据类型']} +TABLES = {'inventory': 'inventory_detail', 'outbound': 'daily_outbound'} +SPECIALS = {'POZ-FM-0184-TM-SM': '.055', 'POZ-XFS-9014-TM-MD': '.062', + 'POZ-FM-9742-TM-MD': '.181', 'POZ-MM-0580-TM-MD': '.22', + **{f'POZ-SKS-{n}-LD': '.1' for n in ('3374', '3367', '3350', '3343')}} +DWS_RETRY_DELAYS = (2, 5, 10) + + +def now(): + return datetime.now(ZoneInfo('Asia/Shanghai')) + + +def command(args, timeout=180, retry_delays=()): + """Run a command, optionally retrying without exposing credential-bearing output.""" + attempts = len(retry_delays) + 1 + failure = 'failed' + for attempt in range(attempts): + try: + result = subprocess.run(args, capture_output=True, timeout=timeout) + if not result.returncode: + return result.stdout + failure = f'failed (exit {result.returncode})' + except subprocess.TimeoutExpired: + failure = f'timed out after {timeout}s' + if attempt < len(retry_delays): + time.sleep(retry_delays[attempt]) + # Never echo command output: authentication errors may contain tokens. + raise RuntimeError(f'{Path(args[0]).name} {failure} after {attempts} attempts') + + +def dws_command(args, timeout=180): + # A single daily job may recover from brief DingDrive/IPaaS failures. This + # is bounded in-process retrying, not polling or an additional schedule. + return command(args, timeout=timeout, retry_delays=DWS_RETRY_DELAYS) + + +def encode(value): + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), default=str).encode() + + +def atomic(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent) + try: + with os.fdopen(fd, 'wb') as out: + out.write(value if isinstance(value, bytes) else encode(value)) + out.flush() + os.fsync(out.fileno()) + os.replace(tmp, path) + finally: + if os.path.exists(tmp): + os.unlink(tmp) + + +def read_json(path, default=None): + return json.loads(path.read_bytes()) if path.exists() else default + + +def clean(v): + return str(v or '').replace('"', '').replace('=', '').strip() + + +def number(v, integer=False): + value = Decimal(clean(v).replace(',', '') or '0') + if not value.is_finite() or (integer and value != value.to_integral_value()): + raise ValueError('Invalid numeric CSV field') + return int(value) if integer else format(value.quantize(Decimal('.0001')), 'f') + + +def bad(code): + return clean(code).upper().rstrip(':：') in {'', 'NA', 'N/A', '合计'} + + +def csv_reader(path): + raw = path.read_bytes() + for encoding in ('utf-8-sig', 'gb18030'): + try: + content = raw.decode(encoding) + break + except UnicodeDecodeError: + continue + else: + raise ValueError('Unrecognized CSV encoding') + reader = csv.DictReader(io.StringIO(content)) + if not reader.fieldnames: + raise ValueError('Missing CSV header') + reader.fieldnames = [clean(x).lstrip('\ufeff') for x in reader.fieldnames] + return reader + + +def pick(row, keys): + return next((clean(row[k]) for k in keys if clean(row.get(k))), '') + + +def parse(path, kind, day, key, products): + reader = csv_reader(path) + headers = set(reader.fieldnames) + if kind == 'inventory': + required = {'货品编号', '货品名称', '成本价', '采购在途', '生产在途量'} + qty_col = '库存' if '库存' in headers else '可用库存' + required.add(qty_col) + elif key == 'wdt': + required = {'货品编号', '货品名称', '货品数量', '货品成本'} + else: + required = set() + if not headers.intersection({'货品编号', '产品编码', '商品编码', '编码', 'product_code', 'goods_code'}): + raise ValueError(f'{path.name}: Missing product code column') + if not headers.intersection({'货品数量', '数量', '出库量', 'qty', 'quantity'}): + raise ValueError(f'{path.name}: Missing quantity column') + if required - headers: + raise ValueError(f'{path.name}: missing columns {sorted(required - headers)}; retained existing data') + rows = [] + valid_codes = 0 + for row in reader: + code = pick(row, ('货品编号', '产品编码', '商品编码', '编码', 'product_code', 'goods_code')) + if bad(code): + continue + valid_codes += 1 + if kind == 'inventory': + name = clean(row.get('货品名称')) + if name in {'-', '--'}: + continue + ct = clean(row.get('单品创建时间')) + created = None + if ct: + created = datetime.fromisoformat(ct).strftime('%Y-%m-%d %H:%M:%S') + rows.append([day, 'wdt', key, clean(row.get('商家编码')), code, name, + clean(row.get('品牌')), number(SPECIALS.get(code, row.get('成本价'))), + number(row.get(qty_col), True), number(row.get('生产在途量'), True), + number(row.get('采购在途'), True), clean(row.get('一级类目')), + clean(row.get('二级类目')), clean(row.get('三级类目')), created]) + else: + if key == 'wdt' and clean(row.get('日期')): + row_day = clean(row['日期'])[:10].replace('/', '-') + if date.fromisoformat(row_day).isoformat() != day: + raise ValueError(f'{path.name}: row date does not match filename') + qty = number(pick(row, ('货品数量', '数量', '出库量', 'qty', 'quantity')), True) + p = row if key == 'wdt' else products.get(code, {}) + name = clean(p.get('货品名称') if key == 'wdt' else p.get('product_name')) + brand = clean(p.get('品牌') if key == 'wdt' else p.get('brand')) + if qty <= 0 or name in {'-', '--'} or '潘思宇代发' in brand: + continue + cats = ['一级类目', '二级类目', '三级类目'] if key == 'wdt' else ['category1', 'category2', 'category3'] + rows.append([code, name or code, brand, clean(row.get('仓库')) if key == 'wdt' else '', + clean(row.get('店铺')) if key == 'wdt' else '', *[clean(p.get(x)) for x in cats], + qty, number(p.get('货品成本') if key == 'wdt' else p.get('cost_price')), + day, key, '出库数据']) + if not rows and (kind == 'inventory' or not valid_codes): + raise ValueError(f'{path.name}: no valid records; refusing replacement') + return rows + + +def connect(local): + spec = importlib.util.spec_from_file_location('sync_config', ROOT / 'config.py') + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + config = dict(mod.DB_CONFIG) + if local: + config.update(user='dy_test', password=(ROOT / '.local-test-db-password').read_text().strip(), + unix_socket='/run/mysqld/mysqld.sock') + config.update(cursorclass=pymysql.cursors.DictCursor, autocommit=False) + return pymysql.connect(**config) + + +def where(packet): + if packet['kind'] == 'inventory': + return 'data_date=%s AND source=%s AND warehouse_code=%s', (packet['date'], 'wdt', packet['key']) + return '`日期`=%s AND `数据来源`=%s', (packet['date'], packet['key']) + + +def normalize(rows, kind): + money_index = 7 if kind == 'inventory' else 9 + result = [] + for row in rows: + r = [str(v) if isinstance(v, (date, datetime)) else v for v in row] + r[money_index] = number(r[money_index]) + result.append(r) + return result + + +def digest(rows, kind): + return hashlib.sha256(b'\n'.join(sorted(encode(r) for r in normalize(rows, kind)))).hexdigest() + + +def db_rows(conn, p): + condition, params = where(p) + with conn.cursor() as cur: + cur.execute('SELECT `' + '`,`'.join(COLS[p['kind']]) + '` FROM ' + TABLES[p['kind']] + ' WHERE ' + condition, params) + return [[r[c] for c in COLS[p['kind']]] for r in cur.fetchall()] + + +def refresh(): + # Load cache modules directly, avoiding services/__init__ and any DB/AI side effects. + for name, function in (('cache', 'refresh_all_caches'), ('simple_cache', 'clear')): + spec = importlib.util.spec_from_file_location('sync_' + name, ROOT / 'services' / (name + '.py')) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + getattr(module, function)() + + +def apply(conn, packets, dry_run=False): + changed = [] + for p in packets: + if p['kind'] not in COLS or p['date'] != date.fromisoformat(p['date']).isoformat(): + raise ValueError('Invalid packet target') + if any(len(r) != len(COLS[p['kind']]) for r in p['rows']): + raise ValueError('Invalid packet shape') + old = db_rows(conn, p) + if digest(old, p['kind']) != digest(p['rows'], p['kind']): + changed.append(p) + if not dry_run: + condition, params = where(p) + with conn.cursor() as cur: + cur.execute('SELECT * FROM ' + TABLES[p['kind']] + ' WHERE ' + condition, params) + atomic(STATE / 'backups' / (now().strftime('%Y%m%dT%H%M%S%f') + '.json.gz'), + gzip.compress(encode({'target': {k: p[k] for k in ('kind', 'date', 'key')}, 'rows': cur.fetchall()}))) + cur.execute('DELETE FROM ' + TABLES[p['kind']] + ' WHERE ' + condition, params) + sql = 'INSERT INTO ' + TABLES[p['kind']] + ' (`' + '`,`'.join(COLS[p['kind']]) + '`) VALUES (' + ','.join(['%s'] * len(COLS[p['kind']])) + ')' + if p['rows']: + cur.executemany(sql, p['rows']) + if digest(db_rows(conn, p), p['kind']) != digest(p['rows'], p['kind']): + raise RuntimeError('Database readback mismatch; transaction rolled back') + if dry_run: + conn.rollback() + else: + conn.commit() + # Refresh even after interrupted commit; publication/ack happens only after refresh succeeds. + if packets: + refresh() + return [{'kind': p['kind'], 'date': p['date'], 'key': p['key'], 'rows': len(p['rows'])} for p in changed] + + +def list_files(kind): + items, cursor, seen = [], None, set() + while True: + args = [DWS, 'drive', 'list', '--space-id', SPACE, '--folder', FOLDERS[kind], '--limit', '50', '--format', 'json'] + if cursor: + args.extend(['--cursor', cursor]) + response = json.loads(dws_command(args)) + if response.get('success') is False or 'result' not in response: + raise RuntimeError('DingDrive list failed') + items.extend(response['result'].get('items', [])) + cursor = response['result'].get('nextToken') + if not cursor: + return items + if cursor in seen: + raise RuntimeError('Repeated DingDrive pagination cursor') + seen.add(cursor) + + +def cloud(conn, args): + config = read_json(STATE / 'settings.json') + target = args.target_date or now().date() + old = read_json(STATE / 'manifest.json', {'entries': {}}) + # Publish only this run's dates. Missing files never become a catch-up queue. + entries = {} + packets, missing = [], [] + with conn.cursor() as cur: + cur.execute('SELECT product_code,product_name,brand,category1,category2,category3,cost_price FROM products') + products = {r['product_code']: r for r in cur.fetchall()} + for kind in FOLDERS: + files = {} + for item in list_files(kind): + m = PATTERNS[kind].fullmatch(item.get('name', '')) + if not m or item.get('type') != 'FILE': + continue + day, key = m.groups() + day = datetime.strptime(day, '%Y%m%d').date() + key = 'dymuto' if key == 'dymtuo' else key + if kind == 'outbound' and key not in SOURCES: + continue + identity = (day, key) + if identity not in files or item['modifyTime'] > files[identity]['modifyTime']: + files[identity] = item + begin = target - timedelta(days=kind == 'outbound') + end = target - timedelta(days=kind == 'outbound') + day = begin + while day <= end: + expected = set(config['warehouses']) if kind == 'inventory' else SOURCES + uploaded = {k for d, k in files if d == day} + for key in sorted(expected - uploaded): + missing.append(f'{kind}:{day}:{key}') + for key in sorted(uploaded): + item = files[(day, key)] + identity = f'{kind}:{day}:{key}' + fingerprint = hashlib.sha256(encode([item['fileId'], item.get('md5'), item['modifyTime'], item['fileSize']])).hexdigest() + existing = old['entries'].get(identity) + if existing and existing['source_fingerprint'] == fingerprint: + entries[identity] = existing + continue + dest = STATE / 'downloads' / item['name'] + dest.parent.mkdir(parents=True, exist_ok=True) + dws_command([DWS, 'drive', 'download', '--node', item['fileId'], '--space-id', SPACE, '--output', str(dest), '--format', 'json']) + if dest.stat().st_size != item['fileSize']: + raise ValueError(f'{item["name"]}: incomplete download') + p = {'kind': kind, 'date': str(day), 'key': key, + 'rows': parse(dest, kind, str(day), key, products)} + blob = gzip.compress(encode(p), mtime=0) + sha = hashlib.sha256(blob).hexdigest() + atomic(STATE / 'packets' / (sha + '.json.gz'), blob) + entries[identity] = {'packet': sha, 'source_fingerprint': fingerprint, + 'rows': len(p['rows']), 'source_name': item['name']} + packets.append(p) + day += timedelta(days=1) + changed = apply(conn, packets, args.dry_run) + if not args.dry_run and not packets and not (STATE / 'manifest.json').exists(): + refresh() + coverage = {kind: sum(identity.startswith(f'{kind}:{target - timedelta(days=kind == "outbound")}:') + for identity in entries) for kind in FOLDERS} + status = {'checked_at': now().isoformat(), 'target_inventory': str(target), + 'target_outbound': str(target - timedelta(days=1)), 'missing': missing, + 'state': 'success', 'missing_policy': 'skip', 'changed': changed, + 'cache_refreshed': not args.dry_run, 'files': len(entries), 'coverage': coverage} + if not args.dry_run: + manifest = {'entries': entries, 'missing': missing, 'missing_policy': 'skip', 'target_inventory': str(target), + 'target_outbound': status['target_outbound'], 'coverage': coverage} + manifest['version'] = hashlib.sha256(encode(manifest)).hexdigest() + status['version'] = manifest['version'] + atomic(STATE / 'status.json', status) + atomic(STATE / 'manifest.json', manifest) + print(json.dumps(status, ensure_ascii=False)) + + +def local(conn, args): + manifest = json.loads(command(SSH + ['cat', REMOTE + '/manifest.json'])) + cloud_status = json.loads(command(SSH + ['cat', REMOTE + '/status.json'])) + target = (args.target_date or now().date()).isoformat() + if manifest.get('target_inventory') != target or manifest.get('missing_policy') != 'skip': + raise RuntimeError('No requested-date single-run publication; other dates will not be imported') + if not cloud_status.get('cache_refreshed') or cloud_status.get('version') != manifest['version']: + raise RuntimeError('Cloud update incomplete or failed; waiting for a validated publication') + old = read_json(STATE / 'local-status.json', {}) + packets = [] + ledger = read_json(STATE / 'local-ledger.json', {}) + for identity, entry in sorted(manifest['entries'].items()): + sha = entry['packet'] + if not re.fullmatch('[a-f0-9]{64}', sha): + raise ValueError('Invalid packet hash') + if ledger.get(identity) == sha: + continue + path = STATE / 'packets' / (sha + '.json.gz') + if not path.exists(): + atomic(path, command(SSH + ['cat', REMOTE + '/packets/' + path.name])) + blob = path.read_bytes() + if hashlib.sha256(blob).hexdigest() != sha: + # Keep the corrupt artifact for diagnosis; let the next attempt fetch a clean copy. + path.rename(path.with_suffix('.corrupt-' + now().strftime('%Y%m%dT%H%M%S%f'))) + raise ValueError('Packet checksum mismatch') + p = json.loads(gzip.decompress(blob)) + if identity != f'{p["kind"]}:{p["date"]}:{p["key"]}': + raise ValueError('Packet identity mismatch') + packets.append(p) + if not packets and old.get('version') == manifest['version'] and old.get('cache_refreshed'): + # Keep online evidence current, without re-invalidating valid caches. + status = {**old, 'checked_at': now().isoformat()} + else: + changed = apply(conn, packets, args.dry_run) + status = {'checked_at': now().isoformat(), 'version': manifest['version'], + 'target_inventory': manifest['target_inventory'], 'target_outbound': manifest['target_outbound'], + 'missing': manifest['missing'], 'state': 'success', 'missing_policy': 'skip', + 'changed': changed, 'cache_refreshed': not args.dry_run, 'files': len(manifest['entries']), + 'coverage': manifest['coverage']} + if not args.dry_run: + atomic(STATE / 'local-ledger.json', {k: v['packet'] for k, v in manifest['entries'].items()}) + atomic(STATE / 'local-status.json', status) + # Exact local readback + cache completion are acknowledged to cloud. + result = subprocess.run(SSH + ['umask 077; tee ' + REMOTE + '/local-ack.json.tmp >/dev/null && mv ' + REMOTE + '/local-ack.json.tmp ' + REMOTE + '/local-ack.json'], input=encode(status), capture_output=True, timeout=45) + if result.returncode: + raise RuntimeError('Cloud acknowledgement failed; no automatic retry') + if old.get('version') != status['version']: + env = dict(os.environ, DBUS_SESSION_BUS_ADDRESS=f'unix:path=/run/user/{os.getuid()}/bus', XDG_RUNTIME_DIR=f'/run/user/{os.getuid()}') + msg = (f"库存 {status['target_inventory']}：{status['coverage']['inventory']}仓库；" + f"出库 {status['target_outbound']}：{status['coverage']['outbound']}来源。两端已同步这些文件并刷新缓存。") + if status['missing']: + msg += f" 本次未上传的{len(status['missing'])}份文件已跳过，不等待补录。" + try: + notification = subprocess.run(['notify-send', '采购数据同步结果', msg], env=env, capture_output=True, timeout=10) + status['desktop_notification'] = 'delivered' if notification.returncode == 0 else 'desktop_unavailable' + except (OSError, subprocess.TimeoutExpired): + status['desktop_notification'] = 'desktop_unavailable' + atomic(STATE / 'local-status.json', status) + print(json.dumps(status, ensure_ascii=False)) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('mode', choices=['cloud', 'local']) + parser.add_argument('--dry-run', action='store_true') + parser.add_argument('--target-date', type=date.fromisoformat, + help='Manual recovery date (YYYY-MM-DD); scheduled runs omit this option') + args = parser.parse_args() + if args.target_date and args.target_date > now().date(): + parser.error('--target-date cannot be in the future') + os.umask(0o077) + STATE.mkdir(parents=True, exist_ok=True) + with open(STATE / (args.mode + '.lock'), 'w') as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print('Already running; skipped') + return + conn = None + try: + conn = connect(args.mode == 'local') + with conn.cursor() as cur: + cur.execute("SELECT GET_LOCK('dy_ding_daily_sync', 0) AS acquired") + if not cur.fetchone()['acquired']: + raise RuntimeError('Another database import is running') + (cloud if args.mode == 'cloud' else local)(conn, args) + except Exception as error: + if conn: + conn.rollback() + failure = {'checked_at': now().isoformat(), 'state': 'error', 'error': str(error), 'cache_refreshed': False} + if not args.dry_run: + atomic(STATE / ('status.json' if args.mode == 'cloud' else 'local-status.json'), failure) + if args.mode == 'local': + # One best-effort failure receipt allows cloud to distinguish failure from offline. + try: + subprocess.run(SSH + ['umask 077; tee ' + REMOTE + '/local-ack.json.tmp >/dev/null && mv ' + REMOTE + '/local-ack.json.tmp ' + REMOTE + '/local-ack.json'], + input=encode(failure), capture_output=True, timeout=30) + except (OSError, subprocess.TimeoutExpired): + pass + print(json.dumps(failure, ensure_ascii=False), file=sys.stderr) + sys.exit(1) + finally: + if conn: + conn.close() + + +if __name__ == '__main__': + main() diff --git a/tests/test_ding_daily_sync_retry.py b/tests/test_ding_daily_sync_retry.py new file mode 100644 index 0000000..0313572 --- /dev/null +++ b/tests/test_ding_daily_sync_retry.py @@ -0,0 +1,59 @@ +import importlib.util +from pathlib import Path +import subprocess +import unittest +from unittest import mock + + +SCRIPT = Path(__file__).resolve().parents[1] / 'scripts' / 'ding_daily_sync.py' +SPEC = importlib.util.spec_from_file_location('ding_daily_sync_retry_test', SCRIPT) +SYNC = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(SYNC) + + +class Result: + def __init__(self, returncode, stdout=b''): + self.returncode = returncode + self.stdout = stdout + + +class DingCommandRetryTest(unittest.TestCase): + def test_dws_command_retries_then_succeeds(self): + results = iter([Result(1), Result(1), Result(0, b'published')]) + sleeps = [] + with mock.patch.object(SYNC.subprocess, 'run', side_effect=lambda *args, **kwargs: next(results)), \ + mock.patch.object(SYNC.time, 'sleep', side_effect=sleeps.append): + self.assertEqual(SYNC.dws_command([SYNC.DWS, 'drive', 'list']), b'published') + self.assertEqual(sleeps, [2, 5]) + + def test_dws_command_stops_after_bounded_attempts(self): + calls = [] + + def run(*args, **kwargs): + calls.append(args) + return Result(1) + + with mock.patch.object(SYNC.subprocess, 'run', side_effect=run), \ + mock.patch.object(SYNC.time, 'sleep'): + with self.assertRaisesRegex(RuntimeError, r'failed \(exit 1\) after 4 attempts'): + SYNC.dws_command([SYNC.DWS, 'drive', 'download']) + self.assertEqual(len(calls), 4) + + def test_dws_command_retries_timeout_without_leaking_output(self): + calls = 0 + + def run(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise subprocess.TimeoutExpired(args[0], kwargs['timeout']) + return Result(0, b'ok') + + with mock.patch.object(SYNC.subprocess, 'run', side_effect=run), \ + mock.patch.object(SYNC.time, 'sleep'): + self.assertEqual(SYNC.dws_command([SYNC.DWS, 'drive', 'list']), b'ok') + self.assertEqual(calls, 2) + + +if __name__ == '__main__': + unittest.main()