#!/usr/bin/env python3 """ Scraper data penelitian (publikasi) dosen dari SINTA. Sumber : halaman authors departemen SINTA -> profil tiap dosen (view googlescholar). Target : tabel `penelitian` pada database dashboard (skema sama dengan crawler_scholar.php lama). Konfigurasi lewat environment variable (default mengikuti config.php dashboard): DB_HOST, DB_USER, DB_PASS, DB_NAME SINTA_USERNAME, SINTA_PASSWORD (opsional; tanpa login SINTA hanya memberi 10 publikasi terbaru per dosen) SINTA_DEPT_URL (opsional; default = departemen Informatika UNTAN) Bisa dijalankan standalone: python sinta_scraper.py --dry-run # scrape saja, tampilkan hasil, tanpa tulis DB python sinta_scraper.py # scrape + sync ke database """ import argparse import logging import os import re import time import requests from bs4 import BeautifulSoup log = logging.getLogger("sinta_scraper") BASE_URL = "https://sinta.kemdiktisaintek.go.id" DEFAULT_DEPT_URL = ( BASE_URL + "/departments/authors/475/EF2A55B7-1CB8-40FC-8D37-B110440E425E/1AD9D155-C081-4913-8B9E-63464F5B0F93" ) USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0 Safari/537.36" ) REQUEST_DELAY = 0.8 # jeda antar request (detik), jangan terlalu agresif ke SINTA MAX_PAGES = 100 # pengaman agar loop pagination tidak lari tanpa batas # ---------------------------------------------------------------- HTTP / login def make_session(): s = requests.Session() s.headers["User-Agent"] = USER_AGENT return s def login(session): """Login ke SINTA bila kredensial tersedia. Return True bila berhasil.""" user = os.getenv("SINTA_USERNAME") pwd = os.getenv("SINTA_PASSWORD") if not user or not pwd: log.warning( "SINTA_USERNAME/SINTA_PASSWORD tidak di-set - tanpa login SINTA hanya " "menampilkan 10 publikasi terbaru per dosen." ) return False session.get(f"{BASE_URL}/logins", timeout=30) # ambil cookie sesi resp = session.post( f"{BASE_URL}/logins/do_login", data={"username": user, "password": pwd}, timeout=30, ) ok = "logout" in resp.text.lower() or "dashboard" in resp.url if ok: log.info("Login SINTA berhasil sebagai %s", user) else: log.warning("Login SINTA GAGAL - lanjut tanpa login (10 publikasi/dosen).") return ok def fetch(session, url): time.sleep(REQUEST_DELAY) resp = session.get(url, timeout=30) resp.raise_for_status() return BeautifulSoup(resp.text, "html.parser") # ------------------------------------------------------------------- scraping def get_department_authors(session, dept_url): """Ambil daftar dosen dari halaman authors departemen (semua halaman). Return list of dict: {sinta_id, nama, gs_id} - gs_id diambil dari URL foto profil (foto SINTA di-serve dari Google Scholar, mengandung parameter user=). """ authors, seen = [], set() for page in range(1, MAX_PAGES): soup = fetch(session, f"{dept_url}?page={page}") found_new = False for name_el in soup.select(".profile-name a[href*='authors/profile/']"): m = re.search(r"authors/profile/(\d+)", name_el["href"]) if not m or m.group(1) in seen: continue seen.add(m.group(1)) found_new = True gs_id = None row = name_el.find_parent("div", class_="col-lg") row = row.parent if row else None if row: img = row.select_one("img") gm = re.search(r"[?&]user=([\w-]+)", img["src"]) if img else None gs_id = gm.group(1) if gm else None authors.append( { "sinta_id": m.group(1), "nama": name_el.get_text(strip=True).title(), "gs_id": gs_id, } ) if not found_new: break log.info("Ditemukan %d dosen di halaman departemen", len(authors)) return authors def parse_publication_items(soup): pubs = [] for item in soup.select(".ar-list-item"): title_el = item.select_one(".ar-title a") if not title_el: continue authors_full = "" for a in item.select(".ar-meta a"): txt = a.get_text(strip=True) if txt.startswith("Authors"): authors_full = re.sub(r"^Authors\s*:\s*", "", txt) break year_el = item.select_one(".ar-year") ym = re.search(r"\d{4}", year_el.get_text()) if year_el else None cited_el = item.select_one(".ar-cited") cm = re.search(r"\d+", cited_el.get_text()) if cited_el else None quart_el = item.select_one(".ar-quartile") # hanya ada di view=scopus qm = re.search(r"Q\d", quart_el.get_text()) if quart_el else None pubs.append( { "judul": title_el.get_text(strip=True), "link": title_el.get("href", ""), "author_full": authors_full, "tahun": ym.group(0) if ym else "", "cited": int(cm.group(0)) if cm else 0, "quartile": qm.group(0) if qm else None, } ) return pubs def get_author_publications(session, sinta_id, view="googlescholar"): """Ambil semua publikasi seorang dosen. Pagination SINTA tanpa login tidak konsisten - kadang halaman berikutnya berisi halaman sebelumnya lagi. Karena itu judul di-dedup, dan loop baru berhenti setelah beberapa halaman berturut-turut tidak memberi judul baru. """ all_pubs, seen, stale_pages = [], set(), 0 for page in range(1, MAX_PAGES): new = [] for _ in range(3): # retry: kadang SINTA menyajikan ulang halaman lama soup = fetch( session, f"{BASE_URL}/authors/profile/{sinta_id}?view={view}&page={page}", ) pubs = parse_publication_items(soup) new = [p for p in pubs if p["judul"] not in seen] if new or not pubs: break if new: stale_pages = 0 all_pubs.extend(new) seen.update(p["judul"] for p in new) else: stale_pages += 1 if stale_pages >= 2: break return all_pubs # ------------------------------------------------------------------- database def db_connect(): import pymysql return pymysql.connect( host=os.getenv("DB_HOST", "localhost"), port=int(os.getenv("DB_PORT", "3306")), user=os.getenv("DB_USER", "root"), password=os.getenv("DB_PASS", "root"), database=os.getenv("DB_NAME", "informatika"), charset="utf8mb4", autocommit=False, ) def ensure_schema(conn): """Tambah kolom penelitian.updated_at bila belum ada (dikelola MySQL: terisi otomatis saat INSERT dan ter-update saat UPDATE). Kolom ditaruh paling akhir agar indeks kolom SELECT * di dashboard PHP tidak bergeser.""" with conn.cursor() as cur: cur.execute( "SELECT COUNT(*) FROM information_schema.columns " "WHERE table_schema=DATABASE() AND table_name='penelitian' " "AND column_name='updated_at'" ) if cur.fetchone()[0] == 0: log.info("Menambahkan kolom penelitian.updated_at") cur.execute( "ALTER TABLE penelitian ADD COLUMN updated_at TIMESTAMP NOT NULL " "DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP" ) conn.commit() def norm_title(t): return re.sub(r"[^a-z0-9]+", "", t.lower()) def norm_name_tokens(name): """Token nama tanpa gelar: buang kata bertitik (Dr., S.Kom., M.T., dst).""" name = name.split(",")[0] tokens = { w.lower() for w in re.findall(r"[A-Za-z]+", name) if len(w) > 1 and w.lower() not in {"dr", "ir", "prof", "st", "mt", "kom"} } return tokens def _norm_name_concat(name): """Nama tanpa gelar sebagai satu string huruf kecil, utk kasus token pecah beda ('Tsana'Uddin' vs 'Tsanauddin').""" return "".join(re.findall(r"[A-Za-z]+", name.split(",")[0])).lower() def match_dosen(nama_sinta, dosen_rows): """Cari baris dosen yang namanya paling cocok (jaccard token >= 0.5, atau string nama gabungan identik).""" target = norm_name_tokens(nama_sinta) target_concat = _norm_name_concat(nama_sinta) best, best_score = None, 0.0 for row in dosen_rows: if target_concat and target_concat == _norm_name_concat(row["nama"] or ""): return row tokens = norm_name_tokens(row["nama"] or "") if not tokens or not target: continue score = len(target & tokens) / len(target | tokens) if score > best_score: best, best_score = row, score return best if best_score >= 0.5 else None def sync_to_db(authors, pubs_per_author): """Upsert hasil scrape ke tabel penelitian + isi kolom dosen.sinta.""" conn = db_connect() ensure_schema(conn) stats = {"inserted": 0, "skipped": 0, "dosen_updated": 0} try: with conn.cursor() as cur: cur.execute("SELECT id_dosen, nama, scholar, sinta FROM dosen") dosen_rows = [ {"id_dosen": r[0], "nama": r[1], "scholar": (r[2] or "").strip(), "sinta": r[3]} for r in cur.fetchall() ] cur.execute("SELECT id_author, judul FROM penelitian") existing = {(r[0], norm_title(r[1] or "")) for r in cur.fetchall()} for author in authors: dosen = match_dosen(author["nama"], dosen_rows) # id_author konsisten dgn data lama: utamakan kolom dosen.scholar # agar publikasi lama & baru menyatu, baru fallback ke ID dari # foto profil SINTA, terakhir ID SINTA itu sendiri id_author = ( (dosen["scholar"] if dosen and dosen["scholar"] else None) or author["gs_id"] or author["sinta_id"] ) if dosen and dosen["sinta"] != author["sinta_id"]: cur.execute( "UPDATE dosen SET sinta=%s WHERE id_dosen=%s", (author["sinta_id"], dosen["id_dosen"]), ) stats["dosen_updated"] += 1 if dosen and not dosen["scholar"] and author["gs_id"]: cur.execute( "UPDATE dosen SET scholar=%s WHERE id_dosen=%s", (author["gs_id"], dosen["id_dosen"]), ) dosen["scholar"] = author["gs_id"] for pub in pubs_per_author.get(author["sinta_id"], []): key = (id_author, norm_title(pub["judul"])) if key in existing: stats["skipped"] += 1 continue cur.execute( "INSERT INTO penelitian " "(judul, author, link, tahun, author_full, id_author, biaya) " "VALUES (%s, %s, %s, %s, %s, %s, '')", ( pub["judul"][:400], author["nama"][:100], pub["link"][:300], pub["tahun"][:4], pub["author_full"][:255], id_author, ), ) existing.add(key) stats["inserted"] += 1 conn.commit() except Exception: conn.rollback() raise finally: conn.close() return stats # ------------------------------------------------------------------ orchestra def run(dry_run=False): """Entry point - dipakai oleh Airflow PythonOperator maupun CLI.""" session = make_session() logged_in = login(session) dept_url = os.getenv("SINTA_DEPT_URL", DEFAULT_DEPT_URL) authors = get_department_authors(session, dept_url) if not authors: raise RuntimeError( "Tidak ada dosen terbaca dari halaman departemen SINTA - " "kemungkinan struktur HTML berubah atau situs tidak bisa diakses." ) pubs_per_author, total = {}, 0 for i, author in enumerate(authors, 1): pubs = get_author_publications(session, author["sinta_id"]) pubs_per_author[author["sinta_id"]] = pubs total += len(pubs) log.info("[%d/%d] %s: %d publikasi", i, len(authors), author["nama"], len(pubs)) log.info("Total %d publikasi dari %d dosen (login=%s)", total, len(authors), logged_in) if dry_run: return {"authors": len(authors), "publications": total, "logged_in": logged_in} stats = sync_to_db(authors, pubs_per_author) log.info( "DB sync: %d baris baru, %d sudah ada, %d dosen di-update kolom sinta", stats["inserted"], stats["skipped"], stats["dosen_updated"], ) return {"authors": len(authors), "publications": total, "logged_in": logged_in, **stats} if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") parser = argparse.ArgumentParser(description="Scrape penelitian dosen dari SINTA") parser.add_argument("--dry-run", action="store_true", help="scrape tanpa menulis ke DB") args = parser.parse_args() print(run(dry_run=args.dry_run))