50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
|
import time
|
|||
|
|
import asyncio
|
|||
|
|
import ntplib
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
class PrecisionTimer:
|
|||
|
|
def __init__(self):
|
|||
|
|
self.offset = 0 # 服务器时间 - 本地时间
|
|||
|
|
|
|||
|
|
def sync_time(self):
|
|||
|
|
"""
|
|||
|
|
同步 NTP 时间，计算偏移量
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
client = ntplib.NTPClient()
|
|||
|
|
response = client.request('pool.ntp.org', version=3)
|
|||
|
|
self.offset = response.tx_time - time.time()
|
|||
|
|
print(f"时间同步完成，偏移量: {self.offset:.3f}s")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"NTP同步失败: {e}，将使用系统时间")
|
|||
|
|
|
|||
|
|
def get_server_time(self):
|
|||
|
|
return time.time() + self.offset
|
|||
|
|
|
|||
|
|
async def wait_until(self, target_time_str):
|
|||
|
|
"""
|
|||
|
|
等待直到目标时间 (格式: 2026-02-01 10:00:00)
|
|||
|
|
"""
|
|||
|
|
target_dt = datetime.strptime(target_time_str, "%Y-%m-%d %H:%M:%S")
|
|||
|
|
target_timestamp = target_dt.timestamp()
|
|||
|
|
|
|||
|
|
print(f"正在等待目标时间: {target_time_str}")
|
|||
|
|
|
|||
|
|
while True:
|
|||
|
|
current_time = self.get_server_time()
|
|||
|
|
remaining = target_timestamp - current_time
|
|||
|
|
|
|||
|
|
if remaining <= 0:
|
|||
|
|
print("目标时间已到！触发抢购！")
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
# 动态调整调整休眠时间以节省 CPU 并保持精度
|
|||
|
|
if remaining > 1:
|
|||
|
|
await asyncio.sleep(remaining - 0.5)
|
|||
|
|
elif remaining > 0.1:
|
|||
|
|
await asyncio.sleep(0.01)
|
|||
|
|
else:
|
|||
|
|
# 最后一刻进入忙等以获取最高精度
|
|||
|
|
pass
|