from __future__ import annotations import json import os import sys import threading import time from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) def _load_json(path: Path) -> dict[str, Any]: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) def _dump_json(path: Path, payload: dict[str, Any]) -> None: tmp_path = path.with_suffix(path.suffix + ".tmp") with open(tmp_path, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) os.replace(tmp_path, path) def _append_log(path: Path, message: str) -> None: with open(path, "a", encoding="utf-8") as handle: handle.write(message) handle.write("\n") def _heartbeat_loop( status_path: Path, status: dict[str, Any], lock: threading.Lock, stop_event: threading.Event, interval_seconds: float = 2.0, ) -> None: while not stop_event.wait(interval_seconds): with lock: if status.get("finished_at") is not None: return status["updated_at"] = time.time() _dump_json(status_path, status) def main(run_dir_arg: str) -> int: from app_cmd.config.BuyConfig import BuyConfig from interface.config import RuntimeOptions from task.buy import Buy run_dir = Path(run_dir_arg) status_path = run_dir / "status.json" result_path = run_dir / "result.json" logs_path = run_dir / "events.log" metadata = _load_json(run_dir / "run.json") config = _load_json(run_dir / "config.json") runtime = RuntimeOptions.from_mapping(_load_json(run_dir / "runtime.json")) existing_status = _load_json(status_path) status = { "ok": True, "run_id": metadata["run_id"], "status": "running", "detail": config.get("detail", metadata["run_id"]), "pid": os.getpid(), "created_at": metadata["created_at"], "started_at": time.time(), "updated_at": time.time(), "finished_at": None, "payment_qr_url": None, "order_id": None, "order_detail_url": None, "payment_code_url": None, "error": None, "last_message": None, "heartbeat_timeout_seconds": existing_status.get("heartbeat_timeout_seconds"), "logs_path": str(logs_path), "result_path": str(result_path), "config_path": str(run_dir / "config.json"), "runtime_path": str(run_dir / "runtime.json"), } _dump_json(status_path, status) status_lock = threading.Lock() heartbeat_stop = threading.Event() heartbeat_thread = threading.Thread( target=_heartbeat_loop, args=(status_path, status, status_lock, heartbeat_stop), daemon=True, name="biliTickerBuy-heartbeat", ) heartbeat_thread.start() buy_job = Buy( config=BuyConfig.from_runtime_options( json.dumps(config, ensure_ascii=False), runtime, show_qrcode=False, ), ) final_status = "completed" try: for event in buy_job.stream(): message = event.message if message is not None: _append_log(logs_path, message) with status_lock: status["updated_at"] = time.time() if message is not None: status["last_message"] = message if "抢票成功" in message: final_status = "succeeded" if "有重复订单" in message: final_status = "duplicate_order" payment_state = event.state for key in ( "payment_qr_url", "order_id", "order_detail_url", "payment_code_url", ): value = getattr(payment_state, key, None) if value not in (None, ""): status[key] = value if message.startswith("PAYMENT_QR_URL="): status["payment_qr_url"] = message.split("=", 1)[1] elif message.startswith("ORDER_ID="): status["order_id"] = message.split("=", 1)[1] elif message.startswith("ORDER_DETAIL_URL="): status["order_detail_url"] = message.split("=", 1)[1] elif message.startswith("PAYMENT_CODE_URL="): status["payment_code_url"] = message.split("=", 1)[1] _dump_json(status_path, status) except BaseException as exc: with status_lock: status["status"] = "failed" status["error"] = repr(exc) status["last_message"] = "RUNNER_EXCEPTION={0!r}".format(exc) status["updated_at"] = time.time() status["finished_at"] = time.time() _append_log(logs_path, "RUNNER_EXCEPTION={0!r}".format(exc)) _dump_json(status_path, status) _dump_json( result_path, { "ok": False, "run_id": metadata["run_id"], "status": "failed", "error": repr(exc), "payment_qr_url": status["payment_qr_url"], "order_id": status["order_id"], "order_detail_url": status["order_detail_url"], "payment_code_url": status["payment_code_url"], "logs_path": str(logs_path), "last_message": status["last_message"], }, ) heartbeat_stop.set() heartbeat_thread.join(timeout=3) return 1 with status_lock: status["status"] = final_status status["updated_at"] = time.time() status["finished_at"] = time.time() _dump_json(status_path, status) _dump_json( result_path, { "ok": True, "run_id": metadata["run_id"], "status": final_status, "payment_qr_url": status["payment_qr_url"], "order_id": status["order_id"], "order_detail_url": status["order_detail_url"], "payment_code_url": status["payment_code_url"], "logs_path": str(logs_path), "last_message": status["last_message"], }, ) heartbeat_stop.set() heartbeat_thread.join(timeout=3) return 0 if __name__ == "__main__": if len(sys.argv) != 2: raise SystemExit("usage: managed_runner.py ") raise SystemExit(main(sys.argv[1]))