sgld

@volvo007 toml 现在应该不需要了

Windows 有一个任务计划程序，里面有一个选项，可以在锁屏场景去执行，你看看符合你要求不

pyinstaller 确实存在一些问题，我内网用的，离线打包基本只能用这个，nuitka 我用不了，没有 wheel

已经撞了一路的包过来了 hhh

盲猜虚拟环境问题

你在虚拟环境里面执行
where python

看看用的是全局 python 还是 venv 中的 python

如果是全局的，那么就 .venv/s.../python.exe -m PyInstaller -F ...

大概就这样。之前遇到过，不行的话重装下虚拟环境试试

@cyaki 是哪位呀

@lxy42 好的好的，我看看这个，🙏感谢资源

用 ai 辅助修改了下代码，然后加了一些 print ，直观的了解了下执行流程

import time
import random
from collections import deque
from itertools import count
import heapq

count_id = count(1)

class Future:
def __init__(self):
self._done = False
self._result = None
self._callbacks = []
self._cancelled = False
self._id = f'Future-{next(count_id)}'

def add_done_callback(self, fn):
if self._done:
fn(self)
else:
self._callbacks.append(fn)

def set_result(self, result):
self._result = result
self._done = True
for cb in self._callbacks:
cb(self)

def __await__(self):
if not self._done:
print(f"Future {self._id} is not done, waiting...")
# 这里的 self 是一个 Future 对象, 需要在调用时使用 await 关键字
yield self
return self._result

class Task(Future):
def __init__(self, coro):
super().__init__()
self.coro = coro
print(f"任务初始化, 任务 ID: {self._id}")
loop.call_soon(self.run)

def run(self):
try:
result = self.coro.send(None)
except StopIteration as e:
"""执行"""
print(f"任务 {self._id} 执行完毕, 结果: {e.value}")
self.set_result(e.value)
else:
if isinstance(result, Future):
result.add_done_callback(self._wakeup)
print(f"Task {self._id} is waiting for Future {result._id}")
def _wakeup(self, future: Future):
"""
This method is called when the future is done.
It schedules the task to run again.
"""
print(f"等待完成, 唤醒任务 {self._id}, 结果: {future._result}")
loop.call_soon(self.run)

class EventLoop:
def __init__(self):
self._ready = deque()
self._scheduled = []
self._stopped = False

def call_soon(self, callback, *args):
self._ready.append((callback, args))

def call_later(self, delay, callback, *args):
heapq.heappush(self._scheduled,
(time.time() + delay, callback, args))

def stop(self):
self._stopped = True

def create_task(self, coro):
return Task(coro)

def run_forever(self):
while not self._stopped:
self.run_once()

def run_once(self):
now = time.time()
while self._scheduled and self._scheduled[0][0] <= now:
_, cb, args = heapq.heappop(self._scheduled)
self._ready.append((cb, args))

num = len(self._ready)
for _ in range(num):

# 取出一个任务, 执行它
cb, args = self._ready.popleft()
print(f"----> 执行 {cb}({args}) ---->")
cb(*args)

async def smallrun():
print("Start smallrun")

# 创建一个 Future 对象
# 代表一个将来的结果, 但是现在还不知道结果是什么
fut = Future()
print(f"Future {fut._id} created")
# 功能模拟 --- 随机延迟, 模拟 IO 操作
# IO 结束以后, 调用 fut.set_result(None)
delay = random.random()
loop.call_later(delay, fut.set_result, None)

await fut
print("End smallrun after", delay)
return delay

async def bigrun():
print("Start bigrun")
delay = await smallrun()
print("End bigrun with", delay*10)
return delay * 10

async def main_task():
print("Main task start")
result = await bigrun()
print("Final result:", result)

if __name__ == "__main__":
loop = EventLoop()
loop.create_task(main_task())

# 2.1 秒后停止事件循环
loop.call_later(2.2, loop.stop)
loop.run_forever()

@Trim21 源码太复杂了，有涉及到很多情况，现在在看上面视频里面 up 写的简易版本的，

@thevita 感谢大佬，我理解一下