1

actuator_test.py

@@ -0,0 +1,39 @@
+def exec_def(cmd_str, obj):
+    import requests
+    json = requests.post("http://172.18.162.173:8000/exec", json={"function_str": cmd_str, "call_str": "exec(obj)", "param_dict": {"obj": obj}}).json()
+    print(json)
+
+
+# ================[ 自定义部分 Start ]===============
+exec_str = """
+def exec(obj):
+    return obj
+"""
+obj = {
+    "mail_type": "其他",
+    "establish_time": None,
+    "registcapi": "0.0",
+    "reccap": "",
+    "if_fest_phone": "是",
+    "industry": None,
+    "stockpercent": None,
+    "subindustry": None,
+    "credit_no": "111000000036931723",
+    "cn_stockname": None,
+    "province": "福建省",
+    "econkind": None,
+    "cn_change": None,
+    "last_day_faren_change": None
+}
+# ================[ 自定义部分 End ]===============
+
+
+exec_def(exec_str, obj)
+
+
+def exec(obj):
+    if 'name3' in obj:
+        obj['name3'] = 1
+    else:
+        obj['name3'] = 2
+    return obj

project/push_user_app.py

@@ -97,6 +97,10 @@ def get_des():
 
 
 def send_message(corp_id, agent_id, secret, userid_list=None, title=None):
+    return send_message2(corp_id, agent_id, secret, userid_list, title, get_des(), "https://memory.tianyunperfect.cn/")
+
+
+def send_message2(corp_id, agent_id, secret, userid_list=None, title=None, des=None, url=None):
     global token
 
     if userid_list is None:
@@ -119,8 +123,8 @@ def send_message(corp_id, agent_id, secret, userid_list=None, title=None):
         "agentid": agent_id,
         "textcard": {
             "title": title,
-            "description": get_des(),
-            "url": "https://memory.tianyunperfect.cn/",
+            "description": des,
+            "url": url,
             "btntxt": "更多"
         },
         "safe": 0,
@@ -131,6 +135,7 @@ def send_message(corp_id, agent_id, secret, userid_list=None, title=None):
     json_str = json.dumps(json_dict)
     response_send = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}",
                                   data=json_str)
+    print(json.loads(response_send.text)['errmsg'])
     return json.loads(response_send.text)['errmsg'] == 'ok'
 
 
@@ -148,7 +153,14 @@ def push_msg(user_id, title):
                                 [str(user_id)], title)}
 
 
+@app.get("/push")
+def push(user_id, title, des, url):
+    return {
+        'success': send_message2("ww582505d35b886cde", "1000002", "yDCwvVOHSvb_43Y3e17mZi4E7hEZ2Z3UDyDpuKxzPsQ",
+                                 [str(user_id)], title, des, url)}
+
+
 if __name__ == '__main__':
-    uvicorn.run(app='push_user_app:app', host="0.0.0.0", port=8087, reload=True, debug=True)
+    uvicorn.run(app='push_user_app:app', host="0.0.0.0", port=9100, reload=True, debug=True)
 
 # uvicorn push_user_app:app --reload --host 0.0.0.0 --port 9100

tmp.py

@@ -1,36 +1,23 @@
-import json
-import os
+def exec_def(cmd_str, obj):
+    import requests
+    print(requests.post("http://172.18.162.173:8000/exec", json={
+        "function_str": cmd_str,
+        "call_str": "exec(obj)",
+        "param_dict": {
+            "obj": obj
+        }
+    }).json())
 
-# 获取 Chrome 书签文件路径
-if os.name == "nt":
-    bookmarks_path = os.path.expanduser("~\\AppData\\Local\\Google\\Chrome\\User Data\\Default\\Bookmarks")
-elif os.name == "posix":
-    bookmarks_path = os.path.expanduser("~/Library/Application Support/Google/Chrome/Default/Bookmarks")
-else:
-    raise NotImplementedError("不支持的操作系统")
 
-# 读取 Chrome 书签文件
-with open(bookmarks_path, "r", encoding="utf-8") as f:
-    bookmarks = json.load(f)
+# ================[ 自定义部分 Start ]===============
+exec_str = """
+def exec(obj):
+    import json
+    obj["a"] = 123
+    return obj
+"""
+obj = {"b": 45}
+# ================[ 自定义部分 End ]===============
 
 
-# test
-# 提取书签的标题和网址
-def extract_bookmarks(bookmarks_dict, bookmarks_list):
-    for item in bookmarks_dict["children"]:
-        if "children" in item:
-            extract_bookmarks(item, bookmarks_list)
-        elif "url" in item:
-            title = item["name"]
-            url = item["url"]
-            bookmarks_list.append({"title": title, "url": url})
-
-
-bookmarks_list = []
-extract_bookmarks(bookmarks["roots"]["bookmark_bar"], bookmarks_list)
-
-# 输出标题和网址
-for bookmark in bookmarks_list:
-    print("标题:", bookmark["title"])
-    print("网址:", bookmark["url"])
-    print("-" * 40)
+exec_def(exec_str, obj)