PythonScirptDemo/sdk.py at main · Cafe-scraper/PythonScirptDemo · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
import grpc
import json
import sdk_pb2
import sdk_pb2_grpc

from google.protobuf import empty_pb2
from typing import TypedDict, List


class CafeSDK:
    _channel = grpc.insecure_channel("127.0.0.1:20086")

    class _ParameterService:
        def __init__(self, channel):
            self.stub = sdk_pb2_grpc.ParameterStub(channel)

        def get_input_json_str(self):
            resp = self.stub.GetInputJSONString(empty_pb2.Empty())
            return resp.jsonString

        def get_input_json_dict(self):
            json_str = self.get_input_json_str()
            return json.loads(json_str) if json_str else {}

    class _ResultService:
        def __init__(self, channel):
            self.stub = sdk_pb2_grpc.ResultStub(channel)

        class TableHeader(TypedDict):
            label: str
            key: str
            format: str

        def set_table_header(self, headers: List[TableHeader]):
            headers = sdk_pb2.TableHeader(headers=headers)
            return self.stub.SetTableHeader(headers)

        def push_data(self, dict_obj):
            json_str = json.dumps(dict_obj, ensure_ascii=False)
            data = sdk_pb2.Data(jsonString=json_str)
            return self.stub.PushData(data)

    class _LogService:
        def __init__(self, channel):
            self.stub = sdk_pb2_grpc.LogStub(channel)

        def debug(self, log: str):
            return self.stub.Debug(sdk_pb2.LogBody(log=log))

        def info(self, log: str):
            return self.stub.Info(sdk_pb2.LogBody(log=log))

        def warn(self, log: str):
            return self.stub.Warn(sdk_pb2.LogBody(log=log))

        def error(self, log: str):
            return self.stub.Error(sdk_pb2.LogBody(log=log))

    Parameter = _ParameterService(_channel)
    Result = _ResultService(_channel)
    Log = _LogService(_channel)


if __name__ == "__main__":
    input_json_str = CafeSDK.Parameter.get_input_json_str()
    print("get_input_json_str res:", input_json_str)

    input_dict = CafeSDK.Parameter.get_input_json_dict()
    print("get_input_json_dict res:", input_dict)

    res = CafeSDK.Result.push_data({"py-data-key": "py-data-value"})
    print("push_data resp:", res.code, res.message)

    res = CafeSDK.Log.debug("py-debug...")
    print("Log.debug resp:", res.code, res.message)

    res = CafeSDK.Log.debug(f"input_dict={input_dict}")
    print("Log.debug resp:", res.code, res.message)

    res = CafeSDK.Log.info("py-info...")
    print("Log.info resp:", res.code, res.message)

    res = CafeSDK.Log.warn("py-warn...")
    print("Log.info resp:", res.code, res.message)

    res = CafeSDK.Log.error("py-error...")
    print("Log.info resp:", res.code, res.message)