social_worker.py

# -*- coding: utf-8 -*-
"""
社媒抓取子进程 worker。

为什么要独立成子进程:
主进程里 akshare 依赖的 py_mini_racer(V8 引擎) 与 Playwright(Chromium) 会争抢
同一块 PartitionAlloc 内存池, 第二次初始化直接 FATAL 崩溃。把 Chromium 抓取放到
独立进程, 主进程只保留 V8, 冲突消除; 且本 worker 优先用装有 Playwright 的系统 python
运行, 浏览器抓取能力不丢。

用法:
  python social_worker.py check        -> {"available": true/false}
  python social_worker.py xueqiu|xhs|douyin  -> 单源结果 JSON
  python social_worker.py all          -> {"xueqiu":{...},"xhs":{...},"douyin":{...}}
结果以 JSON 打印到 stdout, 主进程解析。
"""
import sys, os, json

# 让脚本所在目录可 import social_pw
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))


def _main():
    arg = sys.argv[1] if len(sys.argv) > 1 else "all"
    try:
        import social_pw
    except Exception as e:
        print(json.dumps({"__error__": "no_social_pw: " + str(e)[:120]}, ensure_ascii=False))
        return

    if arg == "check":
        try:
            print(json.dumps({"available": bool(social_pw.pw_available())}))
        except Exception:
            print(json.dumps({"available": False}))
        return

    mapping = {
        "xueqiu": social_pw.pw_xueqiu,
        "xhs": social_pw.pw_xhs,
        "douyin": social_pw.pw_douyin,
    }

    if arg == "all":
        out = {}
        for k, fn in mapping.items():
            try:
                out[k] = fn()
            except Exception as e:
                out[k] = {"status": "fail", "score": None, "count": 0,
                          "updatedAt": "", "note": str(e)[:120]}
        print(json.dumps(out, ensure_ascii=False))
        return

    fn = mapping.get(arg)
    if not fn:
        print(json.dumps({"status": "fail", "score": None, "count": 0,
                          "updatedAt": "", "note": "unknown platform: " + arg}, ensure_ascii=False))
        return
    try:
        print(json.dumps(fn(), ensure_ascii=False))
    except Exception as e:
        print(json.dumps({"status": "fail", "score": None, "count": 0,
                          "updatedAt": "", "note": str(e)[:120]}, ensure_ascii=False))


if __name__ == "__main__":
    _main()