东方财富网-港股公告逻辑修改

base/RedisPPData.py

@@ -125,6 +125,7 @@ def NoticeEnterprise_task():
 # 东方财富网公告 美股 港股
 def NoticeDF():
     cnx, cursor = connectSql()
+    cnx_, cursor_ = cnn11()
     # 获取美股企业
     om_mg_query = "SELECT a.SocialCode From EnterpriseInfo a ,EnterpriseType b WHERE a.SocialCode = b.SocialCode AND b.type=6 AND a.Place=0 AND SecuritiesCode is not null AND SecuritiesCode not LIKE '%.%'"
     cursor.execute(om_mg_query)
@@ -162,6 +163,18 @@ def NoticeDF():
         else:
             continue
 
+    # 获取全部美股企业
+    mg_query = "select * from sys_base_enterprise_ipo where category=7 and securities_code is not null and priority =1"
+    cursor_.execute(mg_query)
+    cnx_.commit()
+    mg_result = cursor_.fetchall()
+    mg_social_list = [item[0] for item in mg_result]
+    print('美股企业=======')
+    for item in mg_social_list:
+        if r.lrem('NoticeEnterprise:mgqy_socialCode_add', 0, item) == 0:
+            r.lpush('NoticeEnterprise:mgqy_socialCode_add', item)
+        else:
+            continue
     # # 获取港股企业
     # om_gg_query = "SELECT a.SocialCode From EnterpriseInfo a ,EnterpriseType b WHERE a.SocialCode = b.SocialCode AND b.type=6 And SecuritiesCode like '%.HK'"
     # cursor.execute(om_gg_query)

comData/noticeReport/东方财富网-公告.py

-import os
+import os
@@ -250,7 +250,7 @@ def GetContent(pdf_url,info_url, pdf_name, social_code, year, pub_time, start_ti
     # print(dic_news)
     # 将相应字段通过kafka传输保存
     try:
-        producer = KafkaProducer(bootstrap_servers=['1.95.3.121:9092'],max_request_size=1024*1024*20)
+        producer = KafkaProducer(bootstrap_servers=['1.95.78.131:9092'],max_request_size=1024*1024*20)
         kafka_result = producer.send("researchReportNoticeTopic",
                                      json.dumps(dic_news, ensure_ascii=False).encode('utf8'))
 
@@ -277,10 +277,10 @@ def GetContent(pdf_url,info_url, pdf_name, social_code, year, pub_time, start_ti
         return False
 
 
-def gonggao_info(dic_info):
-    code = dic_info[3]
-    com_name = dic_info[1]
-    social__code = dic_info[2]
+def gonggao_info(code, com_name, social_code):
+    # code = dic_info[3]
+    # com_name = dic_info[1]
+    # social__code = dic_info[2]
     if 'HK' in code:
         # browser.quit()
         return
@@ -349,7 +349,8 @@ def gonggao_info(dic_info):
                 for n1 in range(0, 3):
                     try:
                         ip = baseCore.get_proxy()
-                        json_2 = requests.get(json_url, proxies=ip,verify=False).json()
+                        # json_2 = requests.get(json_url, proxies=ip,verify=False).json()
+                        json_2 = requests.get(json_url, verify=False).json()
                         break
                     except:
                         if n1 == 2:
@@ -398,7 +399,7 @@ if __name__ =='__main__':
         start_time = time.time()
         # 获取企业信息
         # social_code = baseCore.redicPullData('NoticeEnterprise:mgqy_socialCode_add')
-        social_code = 'ZZSN23030900000316'
+        social_code = 'ZZSN22080900000046'
         if not social_code:
             time.sleep(20)
             continue
@@ -409,12 +410,43 @@ if __name__ =='__main__':
             time.sleep(20)
             continue
         dic_info = baseCore.getInfomation(social_code)
-        count = dic_info[15]
-        code = dic_info[3]
-        com_name = dic_info[1]
+        if dic_info:
+            count = dic_info[15]
+            code = dic_info[3]
+            com_name = dic_info[1]
+        else:
+            # log.info(f'数据库中无该企业{social_code}')
+            sql = f"SELECT * FROM sys_base_enterprise_ipo WHERE social_credit_code = '{social_code}' and category=7 and securities_code is not null and priority=1"
+            cursor.execute(sql)
+            data = cursor.fetchall()
+            if data:
+                pass
+            else:
+                # 数据库中并没有该企业 需要新增
+                pass
+            SecuritiesCode = data[3]
+            SecuritiesShortName = data[4]
+            SecuritiesType = data[5]
+            Category = data[6]
+            Exchange = data[7]
+            sql_baseinfo = f"SELECT * FROM sys_base_enterprise WHERE social_credit_code = '{social_code}'"
+            cursor.execute(sql_baseinfo)
+            data_baseinfo = cursor.fetchone()
+            if data_baseinfo:
+                pass
+            CompanyName = data_baseinfo[3]
+            EnglishName = data_baseinfo[34]
+            countryName = data_baseinfo[39]
+            # 写入数据库
+            insert = "INSERT INTO EnterpriseInfo(CompanyName, SocialCode, SecuritiesCode, SecuritiesShortName, EnglishName, SecuritiesType, Category, Exchange, countryName) VALUES (%s, %s)"
+            cursor_.execute(insert, (CompanyName, social_code, SecuritiesCode, SecuritiesShortName, EnglishName, SecuritiesType, Category, Exchange, countryName))
+            cnx_.commit()
+            com_name = CompanyName
+            code = SecuritiesCode
+
         log.info(f'-----开始处理{com_name}----{social_code}------')
         try:
-            gonggao_info(dic_info)
+            gonggao_info(code, com_name, social_code)
         except:
             time.sleep(30)
             baseCore.rePutIntoR('NoticeEnterprise:mgqy_socialCode', social_code)