from __future__ import annotations
|
|
import argparse
|
import json
|
import os
|
import pathlib
|
import subprocess
|
import sys
|
import time
|
|
PROJECT_DEV = pathlib.Path(__file__).resolve().parents[2]
|
if str(PROJECT_DEV) not in sys.path:
|
sys.path.insert(0, str(PROJECT_DEV))
|
|
|
def record(path: pathlib.Path, role: str) -> None:
|
with path.open("a", encoding="ascii") as target:
|
target.write(json.dumps({"role": role, "pid": os.getpid()}) + "\n")
|
target.flush()
|
|
|
def main() -> int:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--role", choices=("worker", "child", "grandchild", "owner"), required=True)
|
parser.add_argument("--pid-file", type=pathlib.Path, required=True)
|
args = parser.parse_args()
|
record(args.pid_file, args.role)
|
if args.role == "owner":
|
from bili_authenticated_extension.job import WindowsJob, close_handles, open_nul_handles
|
|
nul_in, nul_out, nul_error = open_nul_handles()
|
job = WindowsJob()
|
process = job.launch_suspended(
|
[sys.executable, __file__, "--role", "worker", "--pid-file", str(args.pid_file)],
|
stdin_handle=nul_in,
|
stdout_handle=nul_out,
|
stderr_handle=nul_error,
|
)
|
close_handles(nul_in, nul_out, nul_error)
|
del process
|
time.sleep(300)
|
return 0
|
if args.role == "worker":
|
subprocess.Popen(
|
[sys.executable, __file__, "--role", "child", "--pid-file", str(args.pid_file)],
|
stdin=subprocess.DEVNULL,
|
stdout=subprocess.DEVNULL,
|
stderr=subprocess.DEVNULL,
|
)
|
elif args.role == "child":
|
subprocess.Popen(
|
[sys.executable, __file__, "--role", "grandchild", "--pid-file", str(args.pid_file)],
|
stdin=subprocess.DEVNULL,
|
stdout=subprocess.DEVNULL,
|
stderr=subprocess.DEVNULL,
|
)
|
time.sleep(300)
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
