From 8a9345dfccabaa7235123355912805ad41b192e4 Mon Sep 17 00:00:00 2001
From: JIANG <huarch97@gmail.com>
Date: Wed, 10 Dec 2025 11:06:09 +0800
Subject: [PATCH] =?UTF-8?q?=E8=BF=98=E5=8E=9F=E6=97=A7=E4=BE=9D=E8=B5=96?=
 =?UTF-8?q?=E6=96=87=E4=BB=B6=EF=BC=9B=E4=BD=BF=E7=94=A8=E4=BA=8B=E5=8A=A1?=
 =?UTF-8?q?=20Delete=20And=20Copy=20=E5=8A=A0=E9=80=9F=E6=95=B0=E6=8D=AE?=
 =?UTF-8?q?=E7=9A=84=E5=AD=98=E5=82=A8/=E8=A6=86=E7=9B=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 requirements.txt                | Bin 3358 -> 4090 bytes
 run_server.py                   |  17 +--
 simulation.py                   |   5 +-
 timescaledb/schemas/realtime.py | 228 ++++++++++++++--------------
 timescaledb/schemas/scheme.py   | 260 +++++++++++++++++---------------
 5 files changed, 260 insertions(+), 250 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 0f778260c90ab6a00bf85a327a98ff61cb023084..1db579aa4f211afe81c470c0db23e64bb0ee4587 100644
GIT binary patch
literal 4090
zcmaKvPj4ed5XJk9#FYc5NL=ymj_qs`BqR_>djO=YIC6*_+i@oGpTu@H@m}`2hy!~;
z+AoAS^F{aw!29)^njVK08jUBJuC97j^{Tq&pT9?GscV(CX{39aCaI_AF5T$wP5M&T
zv-Bh#>3)_zP9I5XsnuFLSR3m7al3Y|J)?A|{dKy~b1n@lJ#Qoz(p4HrewcNQCBJT@
z>nzcim42&r(AXTOOX*z;+f>qMI+TunYd6+6dA*nYS*L~e%!TB2w!$u34o}8;b)2rV
zM!cwWA2-NWT3yNJDC=Q2alqf_t;AAj@NX>-cUcP>u+nd2b_pwZ85*jLyUOd|XxCQ0
zt%Nx6VFejtJrVkiB-WB(9qlK&or#Iwj+<x`>9Ekxa-o6DPNn}<yN-sFte*@|<RLqW
zX2~*fK1&bseXdN_*@ut<p%GkYFZf3C-}y88D(9yX4!pR?{%s|<(iKRF<Y9+9@MW80
zIc<?vd3}(zypR{G^tJYr?WvIMggJb!GIraWWyu?%T4wJY8DeL2V45kWQ(epKnK?z?
zkg}FbJ*$p>@Y?H3;oQpOvGAY;459rfV-1Oic7H#6V$HQsEt{y@ruEi(dm%h`iu)v;
zs}9XjP#PgthMkzPqs}<kS<9>c=IBcAVL)LZOgB5O<iWWt*7^-Es4V08PVZkeyoMcM
z^~!a#(i6_9BC*j5(VpdKMovn^J|u71`!+(31iAT;G(4Sm@?Dr3D0}IJFt|s!nw)J|
z7FzFxn~pS>Ul9?<H0s{5pqe+*HqYqr!(GeX(^ts;T+fXzdY!AED7lYd5xKRzp-%4=
zot`AO`i)-Uct%W-hE;T(t$rgXjtjoH4i?f!$1%3>(fM2m%}hE*Lg%h#K8KB^yxs~6
zyuB97$Q1hV&>EJ~y=W1#yRUv1t1GR*#*xqkU!He(5$kyP&>%Ga$U<Nn2`B6g#unB+
zu3KVi_7PM1%v=`{H3DOdswIBJk)Fc|KnGnbD#+p`Xx!nEyp#P|lRNZuQLHjPGJ-y9
z00%OzV|Y)0$7}n6cFP!RcfF??!(%vMw*0^Jjr79Q$Pv=!-@aY<K&;Z&t48KjzI~rg
z+ui1YQSHP*__a*8d7ZUVtK<%hiyT3$`^4dNi2Y^#Nv?xh8oJ$(>V!2u<<V&0C&E|B
zI%me6-qERDtB)hQQ|9O>OcLXa5(&JE^B1jMnX^3vNcZ!ZU|w5#s=8;s;lG#}7LE2{
z=0+vlN3uf|51NzyRMIoW!+p3g!BUZvohpisK^=Gd4~UK7vviOxqY1w@`u{84alU?$
zp62V@#9#E~iJo9BJI;jdNVu4h4s}&}p6LIfBJ;GZ?1S{_yFY$C$Pw`wc#wYn>G!vf
z?SG!$y!+whCvX1z<*(n~zu%XKA3ocuw*y6OtY5NhjlHJE$l+tX!4syATIgXBap>09
zPMSD{W0G_ygNr4i)E?DaIz`lsdm*u*Bhc5E%2(7vUw;RN8);ajtBir&a5L(c>5Cc2
zr>k>bI#rp?IOjYad$x5|M$LhoYDG`M9&;Ug_h&GCG@AoskzJ@R=9wM#-N>-O&2(J$
zyl$Aw#7txh)A!@S6dgTpr)hZ6)O~dJhzGL_CkNc*eDwL?&JeYXjHgm^YOCG-x5Dp;
zmO3FXae9$&Dv#YhQ|T_w7GDCM%Dmd`N}Ie3+r+2LT&{Csf<&zv^cj0DPxu-L=6UM&
z<i0P--1%CUdoocjEZvvHE_`%#lSO)*xnxQB7}z}nJ171w;X7!520qSNEsWgc?;Ot9
zd85fZiJG7ioE5Bl<~1WWLd6_HFDc&&@ICPPI}8nUK}QDd?&=$1<4k6<X1XosyL-7K
z49oE$qFVg_F}>7zvp=09g#}_dY;o|l!1Q7}B}2m;O?yK<n2nQMS)7CLvYduk4%u6=
z!aWAA(!axNf5*CeFn!U{h^{#vUtPvm^tq=no197ZL3Zi<M2;P0mN2KB;jI|t9uZ$t
z`&H?Q%~RbV$Ir|S?%(B3=`*yP9s|WKaNXzJz>J7KCqQ`0d`a~L``jnVTB&#Z^v(N!
E0YgSsqyPW_

delta 851
zcmY*Y&u<b@5T0e-E+E^rHARadYuylI6k!(@Xw;b4kjA424;~tVg;GN664=^sk$5s5
zOe7O8J(x6}^iR>N2aRVBo;5Lg@|(9(V<y0RZ+?8=d^7xZUgutB&vz)KZMp?3({=T!
zl0|8nCGO}R`LkxpsKQa1P4ZTNg9npue|CI7ABS$!9`#XVpvp8PT0;1&*3>HeKx-~_
zX-pkzE58q4gYCqo>6270b1UKKHN-znT;Y$2oBYm9uk0$IIW3QBDl*j1X&yh*?-_zn
z=3jbV^ZDzkg0{}5jZOHbMzK^^4eSH6M?;_rvm#Ctif9+@K00Yv8mjR(BS;A=sTuIy
zg{^{|Dyqs5o$bLwI#gL`96lu#e^@9uu_Pkc5K#v}>;Xr3i%d1P&C6g|Hv>L0bNtCX
zgYGi?!@R^t$*du@0-hyT_<eHOtg4)VH&VP7SDXO1M<Z|uPdLX6JbsmSwSYgTMS-8`
zFPzEb48bh()y(aDpxBSl@rb&xIU`Zr<2RWWU$d6oI&Qv-97CnP(r6wTYhK~D<r?d%
z?sLll)>~_BHAJUF=-Wr-$A7EGbJZ7nT9pTjjvj!?i^VCFm0n!B$V;}%Pg2>b%wQi`
zC1A=lWHU;%gO@JOV{n$098e3~Ls$gwLm<i|I(Tj4J;H5A%{D@Y_{CR(rww0(TNmfz
ze|ZrmgL(3*eZgr#>pBJ4D<4+IZ;cwrKLp;02E1Y4aAoH_#r^<Y%7%;Y-}Y3n$H#VE
L{XfdwOMU+XI?JDC

diff --git a/run_server.py b/run_server.py
index 0e0d57e..958ce34 100644
--- a/run_server.py
+++ b/run_server.py
@@ -8,20 +8,11 @@ if __name__ == "__main__":
     if sys.platform == "win32":
         asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
 
-    # 创建配置对象
-    config = uvicorn.Config(
+    # 用 uvicorn.run 支持 workers 参数
+    uvicorn.run(
         "main:app",
         host="0.0.0.0",
         port=8000,
-        loop="asyncio",  # 强制使用 asyncio 事件循环
-        workers=2,  # 如果需要多进程，可以取消注释
+        workers=2,  # 这里可以设置多进程
+        loop="asyncio",
     )
-    server = uvicorn.Server(config)
-
-    # 创建并设置事件循环
-    if sys.platform == "win32":
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-
-    # 运行服务器
-    asyncio.run(server.serve())
diff --git a/simulation.py b/simulation.py
index 374e322..da7f4bf 100644
--- a/simulation.py
+++ b/simulation.py
@@ -1197,7 +1197,7 @@ def run_simulation(
                 cs.append(valve_status)
                 set_status(name_c, cs)
     # 运行并返回结果
-    result = run_project(name_c)
+    run_project(name_c)
     time_cost_end = time.perf_counter()
     print(
         "{} -- Hydraulic simulation finished, cost time: {:.2f} s.".format(
@@ -1227,6 +1227,7 @@ def run_simulation(
     # print(num_periods_result)
     # print(node_result)
     # 存储
+    starttime = time.time()
     if simulation_type.upper() == "REALTIME":
         TimescaleInternalStorage.store_realtime_simulation(
             node_result, link_result, modify_pattern_start_time
@@ -1240,6 +1241,8 @@ def run_simulation(
             modify_pattern_start_time,
             num_periods_result,
         )
+    endtime = time.time()
+    logging.info("store time: %f", endtime - starttime)
     # 暂不需要再次存储 SCADA 模拟信息
     # TimescaleInternalStorage.fill_scheme_simulation_result_to_SCADA(scheme_Type=scheme_Type, scheme_Name=scheme_Name)
 
diff --git a/timescaledb/schemas/realtime.py b/timescaledb/schemas/realtime.py
index 83e66f0..2cc00f8 100644
--- a/timescaledb/schemas/realtime.py
+++ b/timescaledb/schemas/realtime.py
@@ -12,79 +12,75 @@ class RealtimeRepository:
 
     @staticmethod
     async def insert_links_batch(conn: AsyncConnection, data: List[dict]):
-        """Batch insert for realtime.link_simulation using INSERT for performance."""
+        """Batch insert for realtime.link_simulation using DELETE then COPY for performance."""
         if not data:
             return
-        query = """
-            INSERT INTO realtime.link_simulation (time, id, flow, friction, headloss, quality, reaction, setting, status, velocity)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, id) DO UPDATE SET
-                flow = EXCLUDED.flow,
-                friction = EXCLUDED.friction,
-                headloss = EXCLUDED.headloss,
-                quality = EXCLUDED.quality,
-                reaction = EXCLUDED.reaction,
-                setting = EXCLUDED.setting,
-                status = EXCLUDED.status,
-                velocity = EXCLUDED.velocity
-        """
-        async with conn.cursor() as cur:
-            await cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["id"],
-                        item.get("flow"),
-                        item.get("friction"),
-                        item.get("headloss"),
-                        item.get("quality"),
-                        item.get("reaction"),
-                        item.get("setting"),
-                        item.get("status"),
-                        item.get("velocity"),
-                    )
-                    for item in data
-                ],
-            )
+        
+        # 假设同一批次的数据时间是相同的
+        target_time = data[0]["time"]
+
+        # 使用事务确保原子性
+        async with conn.transaction():
+            async with conn.cursor() as cur:
+                # 1. 先删除该时间点的旧数据
+                await cur.execute(
+                    "DELETE FROM realtime.link_simulation WHERE time = %s",
+                    (target_time,)
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                async with cur.copy(
+                    "COPY realtime.link_simulation (time, id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        await copy.write_row((
+                            item["time"],
+                            item["id"],
+                            item.get("flow"),
+                            item.get("friction"),
+                            item.get("headloss"),
+                            item.get("quality"),
+                            item.get("reaction"),
+                            item.get("setting"),
+                            item.get("status"),
+                            item.get("velocity"),
+                        ))
 
     @staticmethod
     def insert_links_batch_sync(conn: Connection, data: List[dict]):
-        """Batch insert for realtime.link_simulation using INSERT for performance (sync version)."""
+        """Batch insert for realtime.link_simulation using DELETE then COPY for performance (sync version)."""
         if not data:
             return
-        query = """
-            INSERT INTO realtime.link_simulation (time, id, flow, friction, headloss, quality, reaction, setting, status, velocity)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, id) DO UPDATE SET
-                flow = EXCLUDED.flow,
-                friction = EXCLUDED.friction,
-                headloss = EXCLUDED.headloss,
-                quality = EXCLUDED.quality,
-                reaction = EXCLUDED.reaction,
-                setting = EXCLUDED.setting,
-                status = EXCLUDED.status,
-                velocity = EXCLUDED.velocity
-        """
-        with conn.cursor() as cur:
-            cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["id"],
-                        item.get("flow"),
-                        item.get("friction"),
-                        item.get("headloss"),
-                        item.get("quality"),
-                        item.get("reaction"),
-                        item.get("setting"),
-                        item.get("status"),
-                        item.get("velocity"),
-                    )
-                    for item in data
-                ],
-            )
+        
+        # 假设同一批次的数据时间是相同的
+        target_time = data[0]["time"]
+
+        # 使用事务确保原子性
+        with conn.transaction():
+            with conn.cursor() as cur:
+                # 1. 先删除该时间点的旧数据
+                cur.execute(
+                    "DELETE FROM realtime.link_simulation WHERE time = %s",
+                    (target_time,)
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                with cur.copy(
+                    "COPY realtime.link_simulation (time, id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        copy.write_row((
+                            item["time"],
+                            item["id"],
+                            item.get("flow"),
+                            item.get("friction"),
+                            item.get("headloss"),
+                            item.get("quality"),
+                            item.get("reaction"),
+                            item.get("setting"),
+                            item.get("status"),
+                            item.get("velocity"),
+                        ))
 
     @staticmethod
     async def get_link_by_time_range(
@@ -213,59 +209,63 @@ class RealtimeRepository:
     async def insert_nodes_batch(conn: AsyncConnection, data: List[dict]):
         if not data:
             return
-        query = """
-            INSERT INTO realtime.node_simulation (time, id, actual_demand, total_head, pressure, quality)
-            VALUES (%s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, id) DO UPDATE SET
-                actual_demand = EXCLUDED.actual_demand,
-                total_head = EXCLUDED.total_head,
-                pressure = EXCLUDED.pressure,
-                quality = EXCLUDED.quality
-        """
-        async with conn.cursor() as cur:
-            await cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["id"],
-                        item.get("actual_demand"),
-                        item.get("total_head"),
-                        item.get("pressure"),
-                        item.get("quality"),
-                    )
-                    for item in data
-                ],
-            )
+        
+        # 假设同一批次的数据时间是相同的
+        target_time = data[0]["time"]
+
+        # 使用事务确保原子性
+        async with conn.transaction():
+            async with conn.cursor() as cur:
+                # 1. 先删除该时间点的旧数据
+                await cur.execute(
+                    "DELETE FROM realtime.node_simulation WHERE time = %s",
+                    (target_time,)
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                async with cur.copy(
+                    "COPY realtime.node_simulation (time, id, actual_demand, total_head, pressure, quality) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        await copy.write_row((
+                            item["time"],
+                            item["id"],
+                            item.get("actual_demand"),
+                            item.get("total_head"),
+                            item.get("pressure"),
+                            item.get("quality"),
+                        ))
 
     @staticmethod
     def insert_nodes_batch_sync(conn: Connection, data: List[dict]):
         if not data:
             return
-        query = """
-            INSERT INTO realtime.node_simulation (time, id, actual_demand, total_head, pressure, quality)
-            VALUES (%s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, id) DO UPDATE SET
-                actual_demand = EXCLUDED.actual_demand,
-                total_head = EXCLUDED.total_head,
-                pressure = EXCLUDED.pressure,
-                quality = EXCLUDED.quality
-        """
-        with conn.cursor() as cur:
-            cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["id"],
-                        item.get("actual_demand"),
-                        item.get("total_head"),
-                        item.get("pressure"),
-                        item.get("quality"),
-                    )
-                    for item in data
-                ],
-            )
+        
+        # 假设同一批次的数据时间是相同的
+        target_time = data[0]["time"]
+
+        # 使用事务确保原子性
+        with conn.transaction():
+            with conn.cursor() as cur:
+                # 1. 先删除该时间点的旧数据
+                cur.execute(
+                    "DELETE FROM realtime.node_simulation WHERE time = %s",
+                    (target_time,)
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                with cur.copy(
+                    "COPY realtime.node_simulation (time, id, actual_demand, total_head, pressure, quality) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        copy.write_row((
+                            item["time"],
+                            item["id"],
+                            item.get("actual_demand"),
+                            item.get("total_head"),
+                            item.get("pressure"),
+                            item.get("quality"),
+                        ))
 
     @staticmethod
     async def get_node_by_time_range(
diff --git a/timescaledb/schemas/scheme.py b/timescaledb/schemas/scheme.py
index 044ab38..e478cfe 100644
--- a/timescaledb/schemas/scheme.py
+++ b/timescaledb/schemas/scheme.py
@@ -13,83 +13,87 @@ class SchemeRepository:
 
     @staticmethod
     async def insert_links_batch(conn: AsyncConnection, data: List[dict]):
-        """Batch insert for scheme.link_simulation using INSERT for performance."""
+        """Batch insert for scheme.link_simulation using DELETE then COPY for performance."""
         if not data:
             return
-        query = """
-            INSERT INTO scheme.link_simulation (time, scheme_type, scheme_name, id, flow, friction, headloss, quality, reaction, setting, status, velocity)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, scheme_type, scheme_name, id) DO UPDATE SET
-                flow = EXCLUDED.flow,
-                friction = EXCLUDED.friction,
-                headloss = EXCLUDED.headloss,
-                quality = EXCLUDED.quality,
-                reaction = EXCLUDED.reaction,
-                setting = EXCLUDED.setting,
-                status = EXCLUDED.status,
-                velocity = EXCLUDED.velocity
-        """
-        async with conn.cursor() as cur:
-            await cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["scheme_type"],
-                        item["scheme_name"],
-                        item["id"],
-                        item.get("flow"),
-                        item.get("friction"),
-                        item.get("headloss"),
-                        item.get("quality"),
-                        item.get("reaction"),
-                        item.get("setting"),
-                        item.get("status"),
-                        item.get("velocity"),
-                    )
-                    for item in data
-                ],
-            )
+
+        # 假设同一批次的数据时间、scheme_type、scheme_name 是相同的
+        target_time = data[0]["time"]
+        target_scheme_type = data[0]["scheme_type"]
+        target_scheme_name = data[0]["scheme_name"]
+
+        # 使用事务确保原子性
+        async with conn.transaction():
+            async with conn.cursor() as cur:
+                # 1. 先删除该时间点、scheme_type、scheme_name 的旧数据
+                await cur.execute(
+                    "DELETE FROM scheme.link_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s",
+                    (target_time, target_scheme_type, target_scheme_name),
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                async with cur.copy(
+                    "COPY scheme.link_simulation (time, scheme_type, scheme_name, id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        await copy.write_row(
+                            (
+                                item["time"],
+                                item["scheme_type"],
+                                item["scheme_name"],
+                                item["id"],
+                                item.get("flow"),
+                                item.get("friction"),
+                                item.get("headloss"),
+                                item.get("quality"),
+                                item.get("reaction"),
+                                item.get("setting"),
+                                item.get("status"),
+                                item.get("velocity"),
+                            )
+                        )
 
     @staticmethod
     def insert_links_batch_sync(conn: Connection, data: List[dict]):
-        """Batch insert for scheme.link_simulation using INSERT for performance (sync version)."""
+        """Batch insert for scheme.link_simulation using DELETE then COPY for performance (sync version)."""
         if not data:
             return
-        query = """
-            INSERT INTO scheme.link_simulation (time, scheme_type, scheme_name, id, flow, friction, headloss, quality, reaction, setting, status, velocity)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, scheme_type, scheme_name, id) DO UPDATE SET
-                flow = EXCLUDED.flow,
-                friction = EXCLUDED.friction,
-                headloss = EXCLUDED.headloss,
-                quality = EXCLUDED.quality,
-                reaction = EXCLUDED.reaction,
-                setting = EXCLUDED.setting,
-                status = EXCLUDED.status,
-                velocity = EXCLUDED.velocity
-        """
-        with conn.cursor() as cur:
-            cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["scheme_type"],
-                        item["scheme_name"],
-                        item["id"],
-                        item.get("flow"),
-                        item.get("friction"),
-                        item.get("headloss"),
-                        item.get("quality"),
-                        item.get("reaction"),
-                        item.get("setting"),
-                        item.get("status"),
-                        item.get("velocity"),
-                    )
-                    for item in data
-                ],
-            )
+
+        # 假设同一批次的数据时间、scheme_type、scheme_name 是相同的
+        target_time = data[0]["time"]
+        target_scheme_type = data[0]["scheme_type"]
+        target_scheme_name = data[0]["scheme_name"]
+
+        # 使用事务确保原子性
+        with conn.transaction():
+            with conn.cursor() as cur:
+                # 1. 先删除该时间点、scheme_type、scheme_name 的旧数据
+                cur.execute(
+                    "DELETE FROM scheme.link_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s",
+                    (target_time, target_scheme_type, target_scheme_name),
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                with cur.copy(
+                    "COPY scheme.link_simulation (time, scheme_type, scheme_name, id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        copy.write_row(
+                            (
+                                item["time"],
+                                item["scheme_type"],
+                                item["scheme_name"],
+                                item["id"],
+                                item.get("flow"),
+                                item.get("friction"),
+                                item.get("headloss"),
+                                item.get("quality"),
+                                item.get("reaction"),
+                                item.get("setting"),
+                                item.get("status"),
+                                item.get("velocity"),
+                            )
+                        )
 
     @staticmethod
     async def get_link_by_scheme_and_time_range(
@@ -239,63 +243,75 @@ class SchemeRepository:
     async def insert_nodes_batch(conn: AsyncConnection, data: List[dict]):
         if not data:
             return
-        query = """
-            INSERT INTO scheme.node_simulation (time, scheme_type, scheme_name, id, actual_demand, total_head, pressure, quality)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, scheme_type, scheme_name, id) DO UPDATE SET
-                actual_demand = EXCLUDED.actual_demand,
-                total_head = EXCLUDED.total_head,
-                pressure = EXCLUDED.pressure,
-                quality = EXCLUDED.quality
-        """
-        async with conn.cursor() as cur:
-            await cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["scheme_type"],
-                        item["scheme_name"],
-                        item["id"],
-                        item.get("actual_demand"),
-                        item.get("total_head"),
-                        item.get("pressure"),
-                        item.get("quality"),
-                    )
-                    for item in data
-                ],
-            )
+
+        # 假设同一批次的数据时间、scheme_type、scheme_name 是相同的
+        target_time = data[0]["time"]
+        target_scheme_type = data[0]["scheme_type"]
+        target_scheme_name = data[0]["scheme_name"]
+
+        # 使用事务确保原子性
+        async with conn.transaction():
+            async with conn.cursor() as cur:
+                # 1. 先删除该时间点、scheme_type、scheme_name 的旧数据
+                await cur.execute(
+                    "DELETE FROM scheme.node_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s",
+                    (target_time, target_scheme_type, target_scheme_name),
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                async with cur.copy(
+                    "COPY scheme.node_simulation (time, scheme_type, scheme_name, id, actual_demand, total_head, pressure, quality) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        await copy.write_row(
+                            (
+                                item["time"],
+                                item["scheme_type"],
+                                item["scheme_name"],
+                                item["id"],
+                                item.get("actual_demand"),
+                                item.get("total_head"),
+                                item.get("pressure"),
+                                item.get("quality"),
+                            )
+                        )
 
     @staticmethod
     def insert_nodes_batch_sync(conn: Connection, data: List[dict]):
         if not data:
             return
-        query = """
-            INSERT INTO scheme.node_simulation (time, scheme_type, scheme_name, id, actual_demand, total_head, pressure, quality)
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
-            ON CONFLICT (time, scheme_type, scheme_name, id) DO UPDATE SET
-                actual_demand = EXCLUDED.actual_demand,
-                total_head = EXCLUDED.total_head,
-                pressure = EXCLUDED.pressure,
-                quality = EXCLUDED.quality
-        """
-        with conn.cursor() as cur:
-            cur.executemany(
-                query,
-                [
-                    (
-                        item["time"],
-                        item["scheme_type"],
-                        item["scheme_name"],
-                        item["id"],
-                        item.get("actual_demand"),
-                        item.get("total_head"),
-                        item.get("pressure"),
-                        item.get("quality"),
-                    )
-                    for item in data
-                ],
-            )
+
+        # 假设同一批次的数据时间、scheme_type、scheme_name 是相同的
+        target_time = data[0]["time"]
+        target_scheme_type = data[0]["scheme_type"]
+        target_scheme_name = data[0]["scheme_name"]
+
+        # 使用事务确保原子性
+        with conn.transaction():
+            with conn.cursor() as cur:
+                # 1. 先删除该时间点、scheme_type、scheme_name 的旧数据
+                cur.execute(
+                    "DELETE FROM scheme.node_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s",
+                    (target_time, target_scheme_type, target_scheme_name),
+                )
+
+                # 2. 使用 COPY 快速写入新数据
+                with cur.copy(
+                    "COPY scheme.node_simulation (time, scheme_type, scheme_name, id, actual_demand, total_head, pressure, quality) FROM STDIN"
+                ) as copy:
+                    for item in data:
+                        copy.write_row(
+                            (
+                                item["time"],
+                                item["scheme_type"],
+                                item["scheme_name"],
+                                item["id"],
+                                item.get("actual_demand"),
+                                item.get("total_head"),
+                                item.get("pressure"),
+                                item.get("quality"),
+                            )
+                        )
 
     @staticmethod
     async def get_node_by_scheme_and_time_range(