main -> develop

crawling/koreatech_article/article.py

@@ -14,7 +14,13 @@
 from table import replace_table, upload_txt
 from login_v2 import login
 from login import get_jwt_token
-from slack_notice import filter_nas, notice_to_slack
+from slack_notice import (
+    BUS_TIMETABLE_TITLE_PATTERN,
+    COOP_BUSINESS_HOURS_TITLE_PATTERN,
+    LECTURE_REGISTRATION_TITLE_PATTERN,
+    filter_nas,
+    notice_to_slack,
+)
 
 from math import ceil
 from hashlib import sha256
@@ -537,6 +543,8 @@ def get_seg():
 
             articles = []
             bus_articles = []
+            coop_articles = []
+            lecture_articles = []
             new_articles = []
 
             connection = connect_db()
@@ -555,10 +563,24 @@ def get_seg():
 
                 articles.extend(board_articles)
 
-                # 버스 알림
+                # 버스/생협/강의 알림
                 if board.is_notice:
-                    # DB에 없고, 키워드가 들어있는 게시글 필터링
-                    bus_articles.extend(filter_nas(connection, board_articles, keywords={"버스", "bus"}))
+                    # DB에 없고, 알림 조건에 맞는 게시글 필터링
+                    bus_articles.extend(filter_nas(
+                        connection,
+                        board_articles,
+                        title_pattern=BUS_TIMETABLE_TITLE_PATTERN,
+                    ))
+                    coop_articles.extend(filter_nas(
+                        connection,
+                        board_articles,
+                        title_pattern=COOP_BUSINESS_HOURS_TITLE_PATTERN,
+                    ))
+                    lecture_articles.extend(filter_nas(
+                        connection,
+                        board_articles,
+                        title_pattern=LECTURE_REGISTRATION_TITLE_PATTERN,
+                    ))
 
                 new_articles.extend(filter_nas(connection, board_articles))
 
@@ -569,7 +591,20 @@ def get_seg():
         finally:
             try:
                 if bus_articles:
-                    notice_to_slack(bus_articles)
+                    notice_to_slack(bus_articles, "bus")
+
+            except Exception as error:
+                raise error
+            try:
+                if coop_articles:
+                    notice_to_slack(coop_articles, "coop")
+
+            except Exception as error:
+                raise error
+            try:
+                if lecture_articles:
+                    notice_to_slack(lecture_articles, "lecture")
+
             except Exception as error:
                 raise error
             finally:
@@ -595,4 +630,3 @@ def get_seg():
                     raise error
                 finally:
                     connection.close()
-

crawling/koreatech_article/slack_notice.py

@@ -1,11 +1,36 @@
+import json
+import re
 import config
 import requests
-from datetime import date
 
 import urllib3
 import pymysql
 
 
+COOP_BUSINESS_HOURS_TITLE_PATTERN = re.compile(
+    r"^(?=.*(?:생협|생활협동조합))"
+    r"(?=.*사업장\s*운영시간\s*안내)"
+    r"(?=.*(?:\d{2,4}(?:학년도)?\s*[-.]?\s*[12]학기|(?:하계|동계)\s*방학)).*$"
+)
+
+BUS_TIMETABLE_TITLE_PATTERN = re.compile(
+    r"^(?=.*(?:통학|셔틀).*?버스)"
+    r"(?=.*운행(?:계획)?\s*(?:안내|알림))"
+    r"(?=.*(?:20\d{2}학년도\s*[12]학기|"
+    r"(?:20\d{2}(?:학년도|년)?\s*)?(?:하계|동계)\s*"
+    r"(?:계절학기\s*[,·]?\s*)?방학(?:\s*기간)?)).*$"
+)
+
+LECTURE_REGISTRATION_TITLE_PATTERN = re.compile(
+    r"^(?:"
+    r"\[수강신청\]\s*20\d{2}학년도\s*[12]학기\s*"
+    r"(?:(?:정규|예비)\s*)?수강신청\s*안내"
+    r"|\[계절학기\]\s*20\d{2}학년도\s*(?:하계|동계)\s*"
+    r"계절학기\s*(?:\d+차\s*)?수강신청\s*안내"
+    r").*$"
+)
+
+
 def connect_db():
     urllib3.disable_warnings()
     conn = pymysql.connect(host=config.MYSQL_CONFIG['host'],
@@ -17,13 +42,21 @@ def connect_db():
     return conn
 
 
-def filter_nas(connection, nas, keywords=None):
+def filter_nas(connection, nas, keywords=None, exclude_keywords=None, title_pattern=None):
 
     articles = tuple(nas)
 
-    # 키워드가 포함된 게시글 필터링
-    if keywords:
-        articles = (a for a in nas for keyword in keywords if keyword in a.title)
+    # 키워드가 포함되었으며, 제외 키워드가 포함되지 않은 게시글 필터링
+    if title_pattern:
+        articles = (a for a in articles if title_pattern.search(a.title))
+    elif keywords:
+        if exclude_keywords:
+            articles = (
+                a for a in nas
+                if any(keyword in a.title for keyword in keywords) and not any(word in a.title for word in exclude_keywords)
+            )
+        else:
+            articles = (a for a in nas for keyword in keywords if keyword in a.title)
 
     need_notice = []
     sql = f"SELECT COUNT(*) FROM koin.new_koreatech_articles ka JOIN koin.new_articles a on ka.article_id = a.id WHERE a.board_id = %s AND ka.portal_num = %s"
@@ -40,64 +73,92 @@ def filter_nas(connection, nas, keywords=None):
 
 def send_message(body):
     try:
-        url = config.SLACK_CONFIG["url"]
         header = {'Content-type': 'application/json'}
 
         print(body)
 
         # 메세지 전송
-        return requests.post(url, headers=header, json=body)
+        return requests.post(config.SLACK_CONFIG["url"], headers=header, json=body)
 
     except Exception as e:
         print("Slack Message 전송에 실패했습니다.")
         print("에러 내용 : ")
         print(e)
 
 
-def notice_to_slack(articles):
-    blocks = [
-        {
-            "type": "header",
-            "text": {
-                "type": "plain_text",
-                "text": "버스 공지 :Bus:",
-                "emoji": True
-            }
-        },
-        {
-            "type": "rich_text",
-            "elements": [
-                {
-                    "type": "rich_text_list",
-                    "style": "bullet",
-                    "elements": []
-                }
-            ]
-        },
-        {
-            "type": "context",
-            "elements": [
-                {
-                    "type": "plain_text",
-                    "text": f"업데이트: {date.today()}",
-                    "emoji": True
-                }
-            ]
-        }
-    ]
-
-    for art in articles:
-        section = {
-            "type": "rich_text_section",
-            "elements": [
-                {
-                    "type": "link",
-                    "url": art.url,
-                    "text": art.title
+def notice_to_slack(articles, notice_type):
+    notice_functions = {
+        "bus": notice_bus_update,
+        "coop": notice_coop_update,
+        "lecture": notice_lecture_update,
+    }
+
+    for article in articles:
+        notice_functions[notice_type](article.id)
+
+
+def notice_article_update(article_id, notice_name, action_prefix):
+    article_url = f"https://koreatech.in/articles/{article_id}"
+    body = {
+        "text": (
+            f"{notice_name} 공지가 올라왔어요. 게시글 링크: {article_url}\n"
+            "업데이트 작업을 진행할까요?"
+        ),
+        "blocks": [
+            {
+                "type": "section",
+                "text": {
+                    "type": "mrkdwn",
+                    "text": (
+                        f"{notice_name} 공지가 올라왔어요. "
+                        f"<{article_url}|게시글 링크>\n"
+                        "업데이트 작업을 진행할까요?"
+                    )
                 }
-            ]
-        }
+            },
+            {
+                "type": "actions",
+                "elements": [
+                    {
+                        "type": "button",
+                        "text": {
+                            "type": "plain_text",
+                            "text": "예",
+                            "emoji": True
+                        },
+                        "style": "primary",
+                        "action_id": f"{action_prefix}:detected",
+                        "value": json.dumps({"article_id": article_id})
+                    },
+                    {
+                        "type": "button",
+                        "text": {
+                            "type": "plain_text",
+                            "text": "아니요",
+                            "emoji": True
+                        },
+                        "action_id": f"{action_prefix}:detected_ignore",
+                        "value": json.dumps({"article_id": article_id})
+                    }
+                ]
+            }
+        ]
+    }
+
+    response = send_message(body)
+    if response is None:
+        raise RuntimeError("Slack 메시지를 전송하지 못했습니다.")
+
+    response.raise_for_status()
+    return response
+
+def notice_bus_update(article_id):
+    return notice_article_update(article_id, notice_name="버스", action_prefix="bus")
+
+
+def notice_lecture_update(article_id):
+    return notice_article_update(article_id, notice_name="강의", action_prefix="lecture")
 
-        blocks[1]["elements"][0]["elements"].append(section)
 
-    send_message({"blocks": blocks})
+def notice_coop_update(article_id):
+    return notice_article_update(article_id, notice_name="생협", action_prefix="coop")

crawling/koreatech_portal/dining.py

@@ -31,6 +31,7 @@
 redis_client = None
 mysql_connection = None
 KST = pytz.timezone('Asia/Seoul')
+PORTAL_PROGRAM_URL = "https://kut90.koreatech.ac.kr/ssoLogin_ext.jsp?&PGM_ID=CO::CO0998W&locale=ko"
 
 LOGIN_COOKIES = [
     {
@@ -84,10 +85,35 @@ def print_flush(target):
     print(target, flush=True)
 
 
+def mask_cookie(value):
+    if value is None:
+        return "None"
+    value = str(value)
+    if len(value) <= 8:
+        return "*" * len(value)
+    return f"{value[:4]}...{value[-4:]}(len={len(value)})"
+
+
+def print_cookie_log(message, cookies):
+    masked = {name: mask_cookie(value) for name, value in cookies.items()}
+    print_flush(f"{message}: {masked}")
+
+
 # 아우누리 포탈에 식단 정보 요청
 def send_request(login_cookies, eat_date, eat_type, restaurant, campus):
-    headers = {"Content-Type": "text/xml; charset=utf-8"}
-    cookies = {cookie['name']: f"{login_cookies[cookie['name']]};Domain={cookie['domain']}" for cookie in LOGIN_COOKIES}
+    headers = {
+        "Content-Type": "text/xml; charset=utf-8",
+        "Referer": PORTAL_PROGRAM_URL,
+        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36",
+        "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
+        "X-Forwarded-For": config.PORTAL_CONFIG["ip"],
+        "X-Real-IP": config.PORTAL_CONFIG["ip"],
+    }
+    cookies = {cookie['name']: login_cookies[cookie['name']] for cookie in LOGIN_COOKIES}
+    print_cookie_log(
+        f"식단 요청 쿠키 전달 확인 date={eat_date}, type={eat_type}, restaurant={restaurant}, campus={campus}",
+        cookies
+    )
     body = f"""<?xml version="1.0" encoding="UTF-8"?>
     <Root xmlns="http://www.nexacroplatform.com/platform/dataset">
         <Parameters>
@@ -113,15 +139,18 @@ def send_request(login_cookies, eat_date, eat_type, restaurant, campus):
         </Dataset>
     </Root>""".encode("utf-8")
 
-    response = requests.post(
+    response = requests.request(
+        "GET",
         "https://kut90.koreatech.ac.kr/nexacroController.do",
         headers=headers,
         cookies=cookies,
         data=body
     )
+    print_flush(f"식단 요청 응답 status={response.status_code}, content_type={response.headers.get('Content-Type')}")
 
     soup = BeautifulSoup(response.text, 'lxml-xml')
-    if soup.find("Parameter", {"id": "ErrorCode"}).text == '0':
+    error_code = soup.find("Parameter", {"id": "ErrorCode"})
+    if error_code and error_code.text == '0':
         return response
     # 잘못된 쿠키 사용 예외 던지기
     raise ConnectionError("식단 요청 실패")
@@ -158,22 +187,46 @@ def connect_mysql():
 
 
 # 로그인 쿠키 취득
+def extract_login_cookie(cookies):
+    login_cookie = {}
+    for cookie in LOGIN_COOKIES:
+        if hasattr(cookies, "get"):
+            try:
+                value = cookies.get(cookie['name'], domain=cookie['domain'])
+            except TypeError:
+                value = cookies.get(cookie['name'])
+        else:
+            value = None
+        login_cookie[cookie['name']] = value
+    return login_cookie
+
+
+def validate_login_cookie(login_cookie):
+    send_request(login_cookie, datetime.today().strftime("%Y-%m-%d"), "lunch", "한식", "Campus1")
+    return login_cookie
+
+
 def get_cookie():
     # redis 캐시 조회
     global redis_client
     login_cookie = {cookie['name']: redis_client.get(cookie['name']) for cookie in LOGIN_COOKIES}
     for name in login_cookie:
         if not login_cookie[name]: continue
         login_cookie[name] = login_cookie[name].decode("utf-8")
+    print_cookie_log("Redis 포털 쿠키 조회 결과", login_cookie)
     try:
-        send_request(login_cookie, datetime.today().strftime("%Y-%m-%d"), "lunch", "한식", "Campus1")
+        validate_login_cookie(login_cookie)
         return login_cookie
     except ConnectionError:
         pass
 
-    # 아우누리 로그인하여 쿠키 취득
     cookies = portal_login()
-    login_cookie = {cookie['name']: cookies.get(cookie['name'], domain=cookie['domain']) for cookie in LOGIN_COOKIES}
+    login_cookie = extract_login_cookie(cookies)
+    print_cookie_log("포털 로그인 신규 쿠키 취득 결과", login_cookie)
+    if not all(login_cookie.values()):
+        raise ConnectionError("포털 로그인 쿠키 취득 실패")
+
+    validate_login_cookie(login_cookie)
     for cookie in LOGIN_COOKIES:
         redis_client.set(cookie['name'], login_cookie[cookie['name']])
     return login_cookie