diff --git a/app/repo/tech_repo.py b/app/repo/tech_repo.py index 0f83e63..b5d7ca6 100644 --- a/app/repo/tech_repo.py +++ b/app/repo/tech_repo.py @@ -35,8 +35,12 @@ def upsert_daily(rows: list) -> int: payload.append(d) cols = ", ".join(_COLS) + ", created_at, updated_at" vals = ", ".join(f":{c}" for c in _COLS) + ", :ts, :ts" - updates = ", ".join(f"{c} = :{c}" for c in _COLS - if c not in ("data_date", "ts_code")) + ", updated_at = :ts" + # ON DUPLICATE 子句用 VALUES(列) 引用插入值, 不再带绑定参数 (:col)。原因: pymysql 的 + # executemany 对 INSERT ... ON DUPLICATE 做多行合并优化, 只把 VALUES 子句的占位符按行 + # 展开, UPDATE 子句里的 :col (→ %s) 不展开却仍算参数, 5000 行批量时参数错位、SQL 里 + # 留下裸 % 报 1064 (2026-09-11 真机首验抓到; 脱库单测 mock 了落表测不到)。 + updates = ", ".join(f"{c} = VALUES({c})" for c in _COLS + if c not in ("data_date", "ts_code")) + ", updated_at = VALUES(updated_at)" return execute_many( f"INSERT INTO pms_tech_daily ({cols}) VALUES ({vals}) " f"ON DUPLICATE KEY UPDATE {updates}", payload) diff --git a/scripts/test_batch27_units.py b/scripts/test_batch27_units.py index 884f8bf..af57b68 100644 --- a/scripts/test_batch27_units.py +++ b/scripts/test_batch27_units.py @@ -310,6 +310,15 @@ def test_pull_ok(): assert got["rows"][0]["ts_code"] == "600000.SH" +@case("F upsert SQL: ON DUPLICATE 子句用 VALUES() 不带绑定参数(pymysql 批量陷阱, 2026-09-11 真机抓到)") +def test_upsert_odku_no_bind(): + import inspect + from app.repo import tech_repo + src = inspect.getsource(tech_repo.upsert_daily) + assert "= VALUES(" in src, "ON DUPLICATE 子句要用 VALUES(列) 引用插入值" + assert 'f"{c} = :{c}"' not in src, "UPDATE 子句不该再用 :col 绑定参数(pymysql executemany 多行合并会把它错位展开)" + + def main(): ok = 0 for name, fn in RESULTS: