私募基金管理人

MySQLPool.py

@@ -25,7 +25,9 @@ class MySQLPool:
                 cursor.execute(sql, params)
                 conn.commit()
         except Exception as e:
-            print(f"Error executing SQL: {e}")
+            message = f"Error executing SQL: {e}"
+            print(message)
+            Utils.write_log('error.log', message)
             conn.rollback()
         finally:
             conn.close()
@@ -38,7 +40,9 @@ class MySQLPool:
                 cursor.execute(sql, params)
                 return cursor.fetchall()
         except Exception as e:
-            print(f"Error fetching data: {e}")
+            message = f"Error fetching data: {e}"
+            print(message)
+            Utils.write_log('error.log', message)
             return []
         finally:
             conn.close()
@@ -64,7 +68,9 @@ class MySQLPool:
                 cursor.executemany(sql, values)
                 conn.commit()
         except Exception as e:
-            print(f"Error inserting data: {e}")
+            message = f"Error inserting data: {e}"
+            print(message)
+            Utils.write_log('error.log', message)
             conn.rollback()
         finally:
             conn.close()
@@ -87,7 +93,9 @@ class MySQLPool:
                 cursor.execute(sql, values)
                 conn.commit()
         except Exception as e:
-            print(f"Error updating data: {e}")
+            message = f"Error updating data: {e}"
+            print(message)
+            Utils.write_log('error.log', message)
             conn.rollback()
         finally:
             conn.close()

Utils.py

@@ -48,10 +48,10 @@ def get_conn_result(conn, file_name=""):
 
 
 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}")
-
+    this_http_url = f"{http_url}?rand={random_float()}&page={this_page}&size={this_size}"
+    print(f"请求url={this_http_url}")
     conn.request("POST",
-                 f"{http_url}?rand={random_float()}&page={this_page}&size={this_size}",
+                 this_http_url,
                  payload, headers)
 
     data = get_conn_result(conn, file_name).read()
@@ -95,3 +95,10 @@ def get_page_result(http_url, this_page, this_size, payload, headers, conn, save
 
     except json.JSONDecodeError as e:
         print(f"请求 Error decoding JSON: {e}")
+
+
+# 创建一个方法写入日志到根目录下runtime/log/,错误的是 error.log，正确的是 success.log
+def write_log(log_type, log_content):
+    with open(f"runtime/log/{log_type}.log", "a", encoding="utf-8") as f:
+        f.write(log_content + "\n")
+        f.close()

amac/amac_asset_backed_special_plan_api.py

@@ -0,0 +1,45 @@
+import json
+
+import Utils
+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"),
+            'user_tenant_id': item.get("userTenantId"),
+            'product_name': item.get("productName"),
+            'product_code': item.get("productCode"),
+            'org_name': item.get("orgName"),
+            'trustee': item.get("trustee"),
+            'registered_date': item.get("registeredDate"),
+            'fund_found_date': item.get("fundFoundDate"),
+            'fund_due_date': item.get("fundDueDate")
+        }
+
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_asset_backed_special_plan', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 资产支持专项计划公示
+http_url = "/amac-infodisc/api/fund/abs"
+
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
+
+print(f"结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/amac_fund_company_collective_investment_products_api.py

@@ -0,0 +1,42 @@
+import json
+
+import Utils
+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"),
+            'bailor': item.get("bailor"),
+            'register_date': item.get("registerDate"),
+            'register_code': item.get("registerCode"),
+            'name': item.get("name"),
+            'manager': item.get("manager"),
+            'has_classify': item.get("hasClassify")
+        }
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_fund_company_collective_investment_products', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 基金公司及子公司集合资管
+http_url = "/amac-infodisc/api/fund/account"
+
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
+
+print(f"结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/amac_fund_private_investment_funds_api.py

@@ -0,0 +1 @@
+# 基金公司私募投资基金-数据为空，无法知道数据类型

amac/amac_futures_company_collection_product_api.py

@@ -0,0 +1,46 @@
+import json
+
+import Utils
+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"),
+            'mpi_name': item.get("mpiName"),
+            'mpi_product_code': item.get("mpiProductCode"),
+            'aoi_name': item.get("aoiName"),
+            'mpi_trustee': item.get("mpiTrustee"),
+            'mpi_create_date': item.get("mpiCreateDate"),
+            'investment_type': item.get("tzlx"),
+            'is_structured': item.get("sfjgh"),
+            'registered_date': item.get("registeredDate"),
+            'due_date': item.get("dueDate"),
+            'fund_status': item.get("fundStatus")
+        }
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_futures_company_collection_product', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 期货公司集合资管产品
+http_url = "/amac-infodisc/api/pof/futures"
+
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
+
+print(f"结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/amac_private_fund_manager_api.py

@@ -1,12 +1,10 @@
 import json
-import time
 
 import Utils
-from time import sleep
-from http.client import RemoteDisconnected
 from Config import mysql_pool, conn, headers, page, size
 
-print("开始时间(精确到毫秒)：", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), time.time() * 1000)
+start_time = Utils.data_time()
+print("开始时间(精确到毫秒)[私募基金管理人]：", start_time)
 
 
 # 因为这里数据一条人员信息amac_member_user中包含了对应的多个证书，所以需要先循环把证书拿出来
@@ -19,29 +17,34 @@ def savetodb(data):
         # 构建人员信息记录
         this_data = {
             'id': item.get("id"),
-            'manager_id': "",
-            'fund_no': item.get("fundNo", ""),
-            'fund_name': item.get("fundName"),
             'manager_name': item.get("managerName"),
-            'manager_type': item.get("managerType"),
-            'working_state': item.get("workingState"),
-            'put_on_record_date': item.get("putOnRecordDate"),
-            'last_quarter_update': item.get("lastQuarterUpdate"),
-            'is_depute_manage': item.get("isDeputeManage"),
-            'url': item.get("url"),
+            'artificial_person_name': item.get("artificialPersonName"),
+            'register_no': item.get("registerNo"),
             'establish_date': item.get("establishDate"),
-            'manager_url': item.get("managerUrl"),
-            'mandator_name': item.get("mandatorName")
+            'manager_has_product': item.get("managerHasProduct"),
+            'url': item.get("url"),
+            'register_date': item.get("registerDate"),
+            'register_address': item.get("registerAddress"),
+            'register_province': item.get("registerProvince"),
+            'register_city': item.get("registerCity"),
+            'reg_adr_agg': item.get("regAdrAgg"),
+            'office_adr_agg': item.get("officeAdrAgg"),
+            'fund_count': item.get("fundCount"),
+            'paid_in_capital': item.get("paidInCapital"),
+            'subscribed_capital': item.get("subscribedCapital"),
+            'has_special_tips': item.get("hasSpecialTips"),
+            'has_credit_tips': item.get("hasCreditTips"),
+            'reg_coordinate': item.get("regCoordinate"),
+            'office_coordinate': item.get("officeCoordinate"),
+            'office_address': item.get("officeAddress"),
+            'office_province': item.get("officeProvince"),
+            'office_city': item.get("officeCity"),
+            'primary_invest_type': item.get("primaryInvestType"),
+            'fund_type_scale_map': item.get("fundTypeScaleMap"),
+            'member_type': item.get("memberType"),
+            'org_form': item.get("orgForm")
         }
 
-        managers_info = item.get("managersInfo", [])
-
-        # 找到到对应的managerId
-        if managers_info:
-            for info in managers_info:
-                this_data['manager_id'] = info.get("managerId", "")
-                break
-
         person_record.append(this_data)
     # 批量插入
     mysql_pool.insert('amac_private_fund_manager', person_record)
@@ -49,55 +52,13 @@ def savetodb(data):
 
 this_page = page
 this_size = size
-payload = json.dumps({})
-
-http_url = "/amac-infodisc/api/pof/fund"
-
-print(f"请求url={http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}")
-
-conn.request("POST",
-             f"{http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}",
-             payload, headers)
-
-data = Utils.get_conn_result(conn, __file__).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)
+payload = json.dumps({"regiProvinceFsc": "province", "offiProvinceFsc": "province",
+                      "establishDate": {"from": "1900-01-01", "to": "9999-01-01"},
+                      "registerDate": {"from": "1900-01-01", "to": "9999-01-01"}})
 
-    print(f"\n开始插入数据==>{this_page}*{this_size}\n")
-    savetodb(data_to_insert)
+# 私募基金管理人
+http_url = "/amac-infodisc/api/pof/manager/query"
 
-    if json_obj.get("totalPages") > 1:
-        for i in range(1, json_obj.get("totalPages")):
-            this_page = i
-            print(f"请求url={http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}")
-            conn.request("POST",
-                         f"{http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}",
-                         payload, headers)
-            res = conn.getresponse()
-            data = res.read()
-            json_str = data.decode("utf-8")
-            print("JSON String", json_str)
-            try:
-                json_obj_arr = json.loads(json_str)
-                print("获取到总条数为:", 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}")
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
 
-print("结束时间(精确到毫秒)：", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), time.time() * 1000)
+print(f"[私募基金管理人]结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/amac_private_fund_manager_product_api.py

@@ -0,0 +1,103 @@
+import json
+import time
+
+import Utils
+from time import sleep
+from http.client import RemoteDisconnected
+from Config import mysql_pool, conn, headers, page, size
+
+print("开始时间(精确到毫秒)：", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), time.time() * 1000)
+
+
+# 因为这里数据一条人员信息amac_member_user中包含了对应的多个证书，所以需要先循环把证书拿出来
+def savetodb(data):
+    # 判断是否为空，为空则跳过直接返回
+    if data is None: return
+
+    person_record = []
+    for item in data:
+        # 构建人员信息记录
+        this_data = {
+            'id': item.get("id"),
+            'manager_id': "",
+            'fund_no': item.get("fundNo", ""),
+            'fund_name': item.get("fundName"),
+            'manager_name': item.get("managerName"),
+            'manager_type': item.get("managerType"),
+            'working_state': item.get("workingState"),
+            'put_on_record_date': item.get("putOnRecordDate"),
+            'last_quarter_update': item.get("lastQuarterUpdate"),
+            'is_depute_manage': item.get("isDeputeManage"),
+            'url': item.get("url"),
+            'establish_date': item.get("establishDate"),
+            'manager_url': item.get("managerUrl"),
+            'mandator_name': item.get("mandatorName")
+        }
+
+        managers_info = item.get("managersInfo", [])
+
+        # 找到到对应的managerId
+        if managers_info:
+            for info in managers_info:
+                this_data['manager_id'] = info.get("managerId", "")
+                break
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_private_fund_manager_product', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+http_url = "/amac-infodisc/api/pof/fund"
+
+print(f"请求url={http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}")
+
+conn.request("POST",
+             f"{http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}",
+             payload, headers)
+
+data = Utils.get_conn_result(conn, __file__).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:
+        for i in range(1, json_obj.get("totalPages")):
+            this_page = i
+            print(f"请求url={http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}")
+            conn.request("POST",
+                         f"{http_url}?rand={Utils.random_float()}&page={this_page}&size={this_size}",
+                         payload, headers)
+            res = conn.getresponse()
+            data = res.read()
+            json_str = data.decode("utf-8")
+            print("JSON String", json_str)
+            try:
+                json_obj_arr = json.loads(json_str)
+                print("获取到总条数为:", 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}")
+
+print("结束时间(精确到毫秒)：", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), time.time() * 1000)

amac/amac_securities_direct_investment_funds_api.py

@@ -0,0 +1,52 @@
+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"),
+            'name': item.get("name"),
+            'code': item.get("code"),
+            'aoin_name': item.get("aoinName"),
+            'manager_name': item.get("managerName"),
+            'create_date': item.get("createDate"),
+            'submit_date': item.get("submitDate"),
+            'fund_type': item.get("fundType"),
+            'organize_type': item.get("organizeType"),
+            'buy_money': item.get("buyMoney"),
+            'scope': item.get("scope"),
+            'working_state': item.get("workingState"),
+            'is_depute_manage': item.get("isDeputeManage"),
+            'mandator_name': item.get("mandatorName")
+        }
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_securities_direct_investment_funds', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 证券公司直投基金
+http_url = "/amac-infodisc/api/aoin/product"
+
+Utils.get_page_result(http_url, this_page, this_size, payload, headers, conn, savetodb, __file__)
+
+print(f"结束时间(精确到毫秒)： {Utils.data_time()} - {start_time}")

amac/amac_securities_private_investment_funds_api.py

@@ -0,0 +1,50 @@
+import json
+
+import Utils
+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"),
+            'product_id': item.get("productId"),
+            'product_name': item.get("productName"),
+            'product_code': item.get("productCode"),
+            'user_tenant_id': item.get("userTenantId"),
+            'mgr_name': item.get("mgrName"),
+            'found_date': item.get("foundDate"),
+            'registered_date': item.get("registeredDate"),
+            'cast_product': item.get("castProduct"),
+            'fund_type': item.get("fundType"),
+            'org_form': item.get("orgForm"),
+            'fund_status': item.get("fundStatus"),
+            'tuo_guan': item.get("tuoGuan"),
+            'trustee': item.get("trustee"),
+            'delmark': item.get("delmark")
+        }
+
+        person_record.append(this_data)
+    # 批量插入
+    mysql_pool.insert('amac_securities_private_investment_funds', person_record)
+
+
+this_page = page
+this_size = size
+payload = json.dumps({})
+
+# 证券公司私募投资基金
+http_url = "/amac-infodisc/api/pof/subfund"
+
+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

@@ -7,8 +7,16 @@ python_files = [
     # 'member_api.py',  # 会员机构综合查询-会员表
     # 'person_org_registration_api.py',  # 基金从业人员资格注册信息-人员机构注册表
     # 'person_org_registration_user_api.py',  # 基金从业人员资格注册信息-人员机构用户信息表、人员机构用户证书表
-    # 'amac_private_fund_manager_api.py'  # 私募基金管理人基金产品
-    'amac_securities_fund_disclosure_api.py'  # 证券公司集合资管产品公示
+    # 'amac_private_fund_manager_product_api.py'  # 私募基金管理人基金产品
+    # 'amac_securities_fund_disclosure_api.py'  # 证券公司集合资管产品公示
+    # 'amac_securities_direct_investment_funds_api.py'  # 证券公司直投基金
+    # 'amac_securities_private_investment_funds_api.py'  # 证券公司私募投资基金
+    # 'amac_fund_private_investment_funds_api.py'  # 基金公司私募投资基金-[数据为空，无法知道数据类型]
+    # 'amac_fund_company_collective_investment_products_api.py'  # 基金公司及子公司集合资管产品
+    # 'amac_asset_backed_special_plan_api.py'  # 资产支持专项计划公示
+    # 'amac_futures_company_collection_product_api.py'  # 期货公司集合资管产品
+    'amac_private_fund_manager_api.py'  # 私募基金管理人
+
 ]
 
 # 启动所有子进程，并存储它们的引用