Power BI Dev
366 lines
13 KiB
Python
366 lines
13 KiB
Python
#!/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))
|