SQLite 및 Oracle 데이터베이스에 TELEGRAM_URL 업데이트 로직 추가

Author: liante0904

scrap_af_main.py

@@ -4,6 +4,8 @@
 import json
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
 from models.SQLiteManager import SQLiteManager
+from models.OracleManagerSQL import OracleManagerSQL
+
 from models.FirmInfo import FirmInfo
 from modules.LS_0 import LS_detail
 from modules.DBfi_19 import fetch_detailed_url
@@ -15,60 +17,67 @@ async def update_firm_telegram_url_by_date(date_str=None):
     firm_names 배열의 모든 인덱스를 순회하며, telegram_update_required가 True인 경우 TELEGRAM_URL 컬럼을 업데이트합니다.
     """
     print('update_firm_telegram_url_by_date')
-    db = SQLiteManager()
+    sqliteDB = SQLiteManager()
+    oracleDB = OracleManagerSQL()
     all_records = []  # 모든 회사의 레코드를 저장할 리스트
 
-    # firm_names 배열의 모든 인덱스를 순회
     for sec_firm_order in range(len(FirmInfo.firm_names)):
         firm_info = FirmInfo(sec_firm_order=sec_firm_order, article_board_order=0)
 
-        # 회사 이름이 공백이 아니고, telegram_update_required가 True인 경우만 처리
         if firm_info.get_firm_name() and firm_info.telegram_update_required:
-            records = await db.fetch_daily_articles_by_date(firm_info=firm_info, date_str=date_str)
-            print(f"Total records count{len(records)}")
-            # records가 빈 리스트가 아닌 경우에만 처리 진행
+            # === SQLite 처리 ===
+            records = await sqliteDB.fetch_daily_articles_by_date(firm_info=firm_info, date_str=date_str)
+            print(f"[SQLite] Total records count: {len(records)}")
             if records:
-                print(f"Fetched records for SEC_FIRM_ORDER {sec_firm_order}: records count{len(records)}")
+                print(f"[SQLite] Fetched records for SEC_FIRM_ORDER {sec_firm_order}")
                 all_records.extend(records)
 
-                # 조건에 따라 추가 작업 수행
                 if sec_firm_order == 19:
-                    # sec_firm_order가 19인 경우 업데이트 수행
-                    print("Updating TELEGRAM_URL for records with SEC_FIRM_ORDER 19")
-                    update_records = await fetch_detailed_url(records)  # all_records 대신 records 사용
+                    print("Updating TELEGRAM_URL for SEC_FIRM_ORDER 19 in SQLite")
+                    update_records = await fetch_detailed_url(records)
                     for record in update_records:
-                        await db.update_telegram_url(record['id'], record['TELEGRAM_URL'])
-                        print(f"Updated TELEGRAM_URL for id {record['id']} with {record['TELEGRAM_URL']}")
-                
+                        await sqliteDB.update_telegram_url(record['id'], record['TELEGRAM_URL'])
+                        print(f"[SQLite] Updated TELEGRAM_URL for id {record['id']}")
+
                 elif sec_firm_order == 0:
-                    # sec_firm_order가 0인 경우 추가 작업 수행
-                    print("Additional processing for SEC_FIRM_ORDER 0")
-                    print(f"records{len(records)}")
+                    print("Additional processing for SEC_FIRM_ORDER 0 in SQLite")
                     for record in records:
-                        await LS_detail(articles=record, firm_info=firm_info)  # all_records 대신 records 사용
-                        r = await db.update_telegram_url(record['id'], record['TELEGRAM_URL'], record['ARTICLE_TITLE'])
+                        await LS_detail(articles=record, firm_info=firm_info)
+                        r = await sqliteDB.update_telegram_url(record['id'], record['TELEGRAM_URL'], record['ARTICLE_TITLE'])
                         print(r)
 
-                    # update_records = LS_detail(articles=records, firm_info=firm_info)  # all_records 대신 records 사용
-                    
-                    # for record in update_records:
-                    #     await db.update_telegram_url(record['id'], record['TELEGRAM_URL'], record['ARTICLE_TITLE'])
-                    #     print(f"Updated TELEGRAM_URL for id {record['id']} with {record['TELEGRAM_URL']}")
+            # === Oracle 처리 ===
+            oracle_records = await oracleDB.fetch_daily_articles_by_date(firm_info=firm_info, date_str=date_str)
+            print(f"[Oracle] Total records count: {len(oracle_records)}")
+            if oracle_records:
+                print(f"[Oracle] Fetched records for SEC_FIRM_ORDER {sec_firm_order}")
+                all_records.extend(oracle_records)
+
+                if sec_firm_order == 19:
+                    print("Updating TELEGRAM_URL for SEC_FIRM_ORDER 19 in Oracle")
+                    update_records = await fetch_detailed_url(oracle_records)
+                    for record in update_records:
+                        await oracleDB.update_telegram_url(record['id'], record['TELEGRAM_URL'])
+                        print(f"[Oracle] Updated TELEGRAM_URL for id {record['id']}")
+
+                elif sec_firm_order == 0:
+                    print("Additional processing for SEC_FIRM_ORDER 0 in Oracle")
+                    for record in oracle_records:
+                        await LS_detail(articles=record, firm_info=firm_info)
+                        r = await oracleDB.update_telegram_url(record['id'], record['TELEGRAM_URL'], record['ARTICLE_TITLE'])
+                        print(r)
 
-    # 전체 회사들의 레코드가 JSON 리스트로 모임
-    if all_records:  # all_records가 비어 있지 않은 경우에만 JSON 변환 및 출력 수행
+    if all_records:
         json_records = json.dumps(all_records, indent=2)
         print(f"Combined JSON Records size:\n{len(json_records)}")
 
+# 메인 함수
 # 메인 함수
 async def main():
-    # firm_info = FirmInfo(
-    #     sec_firm_order=19,
-    #     article_board_order=0
-    # )
-    # TELEGRAM_URL 업데이트 함수 호출
-    print('?????????????')
+    print('Starting TELEGRAM_URL update for all firms...')
     await update_firm_telegram_url_by_date()
+    print('All updates completed.')
+
 
 if __name__ == "__main__":
     asyncio.run(main())

scrap_main.py

@@ -6,6 +6,8 @@
 from send_error import send_message_to_shell
 
 from models.SQLiteManager import SQLiteManager
+from models.OracleManagerSQL import OracleManagerSQL
+
 from utils.date_util import GetCurrentDate
 
 # business
@@ -126,7 +128,6 @@ async def retry_db_insert_in_memory(db, data, table_name, retries=3, delay=60):
     # 모든 시도 실패 시
     print("모든 DB 삽입 시도가 실패했습니다.")
     return False
-
 async def main():
     try:
         print('===================scrap_send===============')
@@ -179,8 +180,8 @@ async def main():
                 print(f"Running {func.__name__}")
                 data = func()  # 동기 함수 호출
                 if data:
-                    total_data.extend(data)  # 데이터 병합
-                    totalCnt += len(data)  # 카운트 증가
+                    total_data.extend(data)
+                    totalCnt += len(data)
             except Exception as e:
                 logging.error(f"Error in {func.__name__}: {str(e)}", exc_info=True)
                 send_message_to_shell(f"Error in sync function {func.__name__}: {str(e)}")
@@ -192,34 +193,41 @@ async def main():
                 print(f"Running {func.__name__}")
                 data = await func()  # 비동기 함수 호출
                 if data:
-                    total_data.extend(data)  # 데이터 병합
-                    totalCnt += len(data)  # 카운트 증가
+                    total_data.extend(data)
+                    totalCnt += len(data)
             except Exception as e:
                 logging.error(f"Error in {func.__name__}: {str(e)}", exc_info=True)
                 send_message_to_shell(f"Error in async function {func.__name__}: {str(e)}")
 
         print('==============전체 레포트 제공 회사 게시글 조회 완료==============')
 
         if total_data:
-            db = SQLiteManager()
+            sqliteDB = SQLiteManager()
+            oracleDB = OracleManagerSQL()
 
-            # 데이터 삽입 시도
+            # SQLite DB에 데이터 삽입
             try:
-                inserted_count, updated_count = db.insert_json_data_list(total_data, 'data_main_daily_send')
+                inserted_count, updated_count = sqliteDB.insert_json_data_list(total_data, 'data_main_daily_send')
                 print(f"총 {totalCnt}개의 게시글을 스크랩하여.. DB에 Insert 시도합니다.")
                 print(f"총 {inserted_count}개의 새로운 게시글을 DB에 삽입했고, {updated_count}개의 게시글을 업데이트했습니다.")
             except Exception as e:
                 print(f"DB 삽입 중 오류 발생: {str(e)}")
                 logging.error(f"DB Insert Error: {str(e)}", exc_info=True)
-
-                # 메모리에서 데이터를 보관하며 재시도
                 print("DB 삽입 실패, 일정 시간 후 재시도합니다...")
-                success = await retry_db_insert_in_memory(db, total_data, 'data_main_daily_send', retries=3, delay=60)
+                success = await retry_db_insert_in_memory(sqliteDB, total_data, 'data_main_daily_send', retries=3, delay=60)
                 if success:
                     print("DB 재삽입 성공.")
                 else:
                     print("DB 재삽입 실패. 데이터를 확인하세요.")
 
+            # Oracle DB에 데이터 삽입
+            try:
+                oracle_inserted_count, oracle_updated_count = oracleDB.insert_json_data_list(total_data, 'data_main_daily_send')
+                print(f"[Oracle] 총 {oracle_inserted_count}개의 새로운 게시글을 삽입했고, {oracle_updated_count}개의 게시글을 업데이트했습니다.")
+            except Exception as e:
+                print(f"[Oracle] DB 삽입 중 오류 발생: {str(e)}")
+                logging.error(f"[Oracle] DB Insert Error: {str(e)}", exc_info=True)
+
             if inserted_count or updated_count:
                 # 추가 비동기 작업 실행
                 await scrap_af_main.main()
@@ -228,9 +236,8 @@ async def main():
         else:
             print("새로운 게시글 스크랩 실패.")
     except Exception as e:
-        # 전체 프로세스 에러 처리
         logging.error("An error occurred in the main process", exc_info=True)
         send_message_to_shell(f"Error in main: {str(e)}")
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())