证券公司集合资管产品公示

Utils.py

@@ -1,9 +1,12 @@
+import json
 import re
 import random
 import time
 from datetime import datetime
 from http.client import RemoteDisconnected
 
+import callback
+
 
 def camel_to_snake(name: str) -> str:
     # 匹配大写字母，并在前面加上下划线，然后转换为小写
@@ -42,3 +45,53 @@ def get_conn_result(conn, file_name=""):
             else:
                 print(f"{file_name} - {data_time()}请求失败，达到最大重试次数")
                 raise
+
+
+def get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb: callback, file_name=""):
+    print(f"请求url={http_url}?rand={random_float()}&page={this_page}&size={this_size}")
+
+    conn.request("POST",
+                 f"{http_url}?rand={random_float()}&page={this_page}&size={this_size}",
+                 payload, headers)
+
+    data = get_conn_result(conn, file_name).read()
+    json_str = data.decode("utf-8")
+
+    try:
+        json_obj = json.loads(json_str)
+        print("获取到总条数为:", json_obj.get("totalPages"))
+
+        data_to_insert = json_obj.get("content")
+
+        # if data_to_insert is not None:
+        #     # 先睡一秒，确保峰值不会太高
+        #     sleep(1)
+
+        print(f"\n开始插入数据==>{this_page}*{this_size}\n")
+        savetodb(data_to_insert)
+
+        if json_obj.get("totalPages") > 1:
+
+            try:
+                for i in range(1, json_obj.get("totalPages")):
+                    this_page = i
+                    print(f"请求url={http_url}?rand={random_float()}&page={this_page}&size={this_size}")
+                    conn.request("POST",
+                                 f"{http_url}?rand={random_float()}&page={this_page}&size={this_size}",
+                                 payload, headers)
+                    data = get_conn_result(conn, file_name).read()
+                    json_str = data.decode("utf-8")
+                    print("JSON String", json_str)
+                    json_obj_arr = json.loads(json_str)
+                    print(f"获取到总页数为:{json_obj_arr.get("totalPages")}页")
+                    data_to_insert = json_obj_arr.get("content")
+                    # if data_to_insert is not None:
+                    #     # 先睡一秒，确保峰值不会太高
+                    #     time.sleep(1)
+                    print(f"\n开始插入数据==>{i}页 * {this_size}\n")
+                    savetodb(data_to_insert)
+            except json.JSONDecodeError as e:
+                print(f"循环 Error decoding JSON: {e}")
+
+    except json.JSONDecodeError as e:
+        print(f"请求 Error decoding JSON: {e}")

amac/amac_securities_fund_disclosure_api.py

@@ -0,0 +1,49 @@
+import json
+import time
+
+import Utils
+from time import sleep
+from http.client import RemoteDisconnected
+from Config import mysql_pool, conn, headers, page, size
+
+start_time = Utils.data_time()
+print("开始时间(精确到毫秒)[证券公司集合资管产品公示]：", start_time)
+
+
+# 因为这里数据一条人员信息amac_member_user中包含了对应的多个证书，所以需要先循环把证书拿出来
+def savetodb(data):
+    # 判断是否为空，为空则跳过直接返回
+    if data is None: return
+
+    person_record = []
+    for item in data:
+        # 构建人员信息记录
+        this_data = {
+            'id': item.get("id"),
+            'cpmc': item.get("cpmc"),
+            'cpbm': item.get("cpbm", ""),
+            'gljg': item.get("gljg"),
+            'slrq': item.get("slrq"),
+            'dqr': item.get("dqr"),
+            'tzlx': item.get("tzlx"),
+            'sffj': item.get("sffj"),
+            'tgjg': item.get("tgjg"),
+            'barq': item.get("barq"),
+            'yzzt': item.get("yzzt")
+        }
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_securities_fund_disclosure ', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 证券公司集合资管产品公示
+http_url = "/amac-infodisc/api/pof/securities"
+
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
+
+print(f"结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/main.py

@@ -6,8 +6,9 @@ print("执行 开始")
 python_files = [
     # 'member_api.py',  # 会员机构综合查询-会员表
     # 'person_org_registration_api.py',  # 基金从业人员资格注册信息-人员机构注册表
-    'person_org_registration_user_api.py',  # 基金从业人员资格注册信息-人员机构用户信息表、人员机构用户证书表
-    'amac_private_fund_manager_api.py'  # 私募基金管理人基金产品
+    # 'person_org_registration_user_api.py',  # 基金从业人员资格注册信息-人员机构用户信息表、人员机构用户证书表
+    # 'amac_private_fund_manager_api.py'  # 私募基金管理人基金产品
+    'amac_securities_fund_disclosure_api.py'  # 证券公司集合资管产品公示
 ]
 
 # 启动所有子进程，并存储它们的引用

amac/person_org_registration_user_api.py

@@ -70,9 +70,8 @@ def savetodb(data):
 
 
 # 开始获取amac_person_org_registration表中的数据并进行数据抓取
-# amac_org_type = mysql_pool.fetchall("select * from amac_person_org_registration")
-amac_org_type = mysql_pool.fetchall("select * from amac_person_org_registration where user_id >= 2201051041561049 "
-                                    "ORDER BY user_id asc")
+amac_org_type = mysql_pool.fetchall("select * from amac_person_org_registration")
+# amac_org_type = mysql_pool.fetchall("select * from amac_person_org_registration where user_id >= 2201051041561049 ORDER BY user_id asc")
 # amac_org_type = [{"user_id": "2309140927320704"}]
 
 for amac_org_type_item in amac_org_type: