Spaces:
Sleeping
Sleeping
| """ | |
| scraper.py | |
| ดึงประกาศจัดซื้อจัดจ้างจากระบบ e-GP ของ มทส. (mis.sut.ac.th) | |
| โครงสร้างจริง (ยืนยันจาก raw HTML ของผู้ใช้): | |
| - Table ข้อมูล: id="myTable", แถวข้อมูล class="LViewer_Row0"/"LViewer_Row1" | |
| คอลัมน์: ลำดับ, เลขที่โครงการ, โค้ดวิธีจัดซื้อ, รายละเอียด, วันที่ประกาศ | |
| มี fileId ฝังใน onclick="goLink('...fileId=XXXX',true)" -> ลิงก์เอกสารประกาศจริง | |
| - Pagination: เป็น ASP.NET WebForm postback (ไม่ใช่ query param ธรรมดา) | |
| form method="get" action="getEGP.aspx" id="aspnetForm" | |
| ต้องส่ง __VIEWSTATE, __VIEWSTATEGENERATOR (อ่านจาก response ก่อนหน้า) กลับไปด้วย | |
| พร้อม FrmCarHdr_PagingMove = '>' (หน้าถัดไป) / '<' (ก่อนหน้า) / '<<' (หน้าแรก) / '>>' (หน้าสุดท้าย) | |
| ทั้งหมดมี 21 หน้า (ตามที่ระบุในตัวเว็บ "หน้าที่ 1 จาก 21 หน้า") | |
| """ | |
| import re | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from dataclasses import dataclass, asdict | |
| from typing import List, Optional | |
| import time | |
| BASE_URL = "https://mis.sut.ac.th/MisPublic/Modules/WebService/getEGP.aspx" | |
| HEADERS = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " | |
| "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36" | |
| } | |
| TOTAL_PAGES = 21 # ยืนยันจาก raw HTML: "หน้าที่ ... จาก 21 หน้า" | |
| REQUEST_DELAY_SEC = 0.5 # หน่วงเวลาระหว่าง request เพื่อไม่โหลดเซิร์ฟเวอร์เขามากเกินไป | |
| class TenderItem: | |
| seq: str # ลำดับ | |
| project_code: str # เลขที่โครงการ (เช่น 69069537783) | |
| method_code: str # โค้ดวิธีจัดซื้อ (เช่น 15, B0, D0 ... ดูคำอธิบายโค้ดด้านบนตาราง) | |
| title: str # รายละเอียดโครงการ (มีคำว่าวิธีจัดซื้อ เช่น e-bidding ปนอยู่ในข้อความนี้) | |
| announce_date: str # วันที่ประกาศ (พ.ศ., เช่น "29 มิ.ย. 69") | |
| file_url: str # ลิงก์ดาวน์โหลดเอกสารประกาศ (ดึงจาก onclick="goLink(...)") | |
| def fetch_html(extra_params: Optional[dict] = None) -> str: | |
| """GET request ไปที่หน้า e-GP พร้อม encoding cp874 ที่ถูกต้อง""" | |
| params = {"TotalListView": "yes"} | |
| if extra_params: | |
| params.update(extra_params) | |
| resp = requests.get(BASE_URL, params=params, headers=HEADERS, timeout=30) | |
| resp.encoding = "cp874" # สำคัญมาก: ไม่ใส่จะได้ภาษาไทยเพี้ยนเป็น "?" | |
| return resp.text | |
| def extract_viewstate(html: str) -> dict: | |
| """ดึง __VIEWSTATE และ __VIEWSTATEGENERATOR จาก HTML เพื่อใช้ในการเปลี่ยนหน้าถัดไป""" | |
| soup = BeautifulSoup(html, "html.parser") | |
| viewstate_tag = soup.find("input", {"id": "__VIEWSTATE"}) | |
| viewstategen_tag = soup.find("input", {"id": "__VIEWSTATEGENERATOR"}) | |
| return { | |
| "__VIEWSTATE": viewstate_tag.get("value", "") if viewstate_tag else "", | |
| "__VIEWSTATEGENERATOR": viewstategen_tag.get("value", "") if viewstategen_tag else "", | |
| } | |
| def parse_row(row) -> Optional[TenderItem]: | |
| """ | |
| Parse แถวข้อมูลจริงจากตาราง id="myTable" | |
| ข้ามแถว header (class="Lister_RowHeader") และแถวที่ไม่ใช่ข้อมูล | |
| """ | |
| row_class = row.get("class", []) | |
| if "Lister_RowHeader" in row_class: | |
| return None # ข้ามแถว header | |
| if not any(c.startswith("LViewer_Row") for c in row_class): | |
| return None # ไม่ใช่แถวข้อมูล (เช่น แถว pagination ด้านล่างตาราง) | |
| cells = row.find_all("td") | |
| if len(cells) < 5: | |
| return None | |
| # ดึง fileId/URL จาก onclick="goLink('...',true)" | |
| onclick = row.get("onclick", "") | |
| file_url = "" | |
| if "goLink(" in onclick: | |
| match = re.search(r"goLink\('([^']+)'", onclick) | |
| if match: | |
| file_url = match.group(1) | |
| return TenderItem( | |
| seq=cells[0].get_text(strip=True), | |
| project_code=cells[1].get_text(strip=True), | |
| method_code=cells[2].get_text(strip=True), | |
| title=cells[3].get_text(strip=True), | |
| announce_date=cells[4].get_text(strip=True), | |
| file_url=file_url, | |
| ) | |
| def parse_page(html: str) -> List[TenderItem]: | |
| soup = BeautifulSoup(html, "html.parser") | |
| table = soup.find("table", {"id": "myTable"}) | |
| if table is None: | |
| return [] | |
| rows = table.find_all("tr") | |
| items = [parse_row(r) for r in rows] | |
| return [i for i in items if i is not None] | |
| def scrape_all(max_pages: int = TOTAL_PAGES, keyword_filter: Optional[str] = None) -> List[dict]: | |
| """ | |
| ดึงทุกหน้าโดยจำลอง ASP.NET postback pagination: | |
| หน้าแรก -> GET ปกติ | |
| หน้าถัดไป -> GET พร้อม __VIEWSTATE/__VIEWSTATEGENERATOR ของหน้าก่อน + FrmCarHdr_PagingMove='>' | |
| """ | |
| results: List[TenderItem] = [] | |
| html = fetch_html() | |
| results.extend(parse_page(html)) | |
| for page_num in range(2, max_pages + 1): | |
| state = extract_viewstate(html) | |
| if not state["__VIEWSTATE"]: | |
| # ดึง viewstate ไม่ได้ -> หยุด ไม่งั้น server จะปฏิเสธ/วนซ้ำหน้าเดิม | |
| break | |
| params = { | |
| "__VIEWSTATE": state["__VIEWSTATE"], | |
| "__VIEWSTATEGENERATOR": state["__VIEWSTATEGENERATOR"], | |
| "FrmCarHdr_PagingMove": ">", | |
| } | |
| time.sleep(REQUEST_DELAY_SEC) | |
| html = fetch_html(params) | |
| page_items = parse_page(html) | |
| if not page_items: | |
| break # ไม่มีข้อมูลเพิ่มแล้ว | |
| results.extend(page_items) | |
| if keyword_filter: | |
| kw = keyword_filter.lower() | |
| results = [r for r in results if kw in r.title.lower()] | |
| return [asdict(r) for r in results] | |
| if __name__ == "__main__": | |
| # ทดสอบดึงแค่ 3 หน้าแรกก่อน (กัน spam เซิร์ฟเวอร์เขาตอน dev) | |
| data = scrape_all(max_pages=3) | |
| print(f"ดึงได้ {len(data)} รายการ") | |
| for d in data[:5]: | |
| print(d) | |