Problem
DMO (Devlet Malzeme Ofisi) tedarikçi portalında 5900+ aktif sipariş takip ediliyordu. Portala manuel giriş yapıp sipariş durumlarını kontrol etmek saatlerce sürüyordu. Ayrıca durum değişikliklerinin SQL Server'a yazılması ve alarm üretilmesi gerekiyordu.
Playwright + CDP Yaklaşımı
Doğrudan Playwright ile browser kontrolü yapmak yerine, kullanıcının mevcut Chrome oturumuna CDP (Chrome DevTools Protocol) üzerinden bağlandım. Bu sayede login işlemini atlıyoruz ve mevcut session'ı kullanıyoruz.
from playwright.sync_api import sync_playwright
def connect_existing_chrome():
"""Açık Chrome oturumuna bağlan (Chrome --remote-debugging-port=9222 ile açılmalı)"""
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp("http://localhost:9222")
context = browser.contexts[0]
page = context.pages[0]
return browser, page
def safe_navigate(page, url):
"""Form uyarısını atlayan navigate — page.goto() yerine"""
page.evaluate(f"window.location.href = '{url}'")
page.wait_for_load_state('networkidle', timeout=30000)
Cookie aktarımı için Playwright session'ından requests kütüphanesine cookie'leri aktardım. Bu sayede ağır Playwright yerine hafif HTTP istekleri kullanılıyor.
import requests
def transfer_cookies(page) -> requests.Session:
"""Playwright cookie'lerini requests Session'a aktar"""
session = requests.Session()
for cookie in page.context.cookies():
session.cookies.set(
cookie['name'], cookie['value'],
domain=cookie.get('domain', ''),
path=cookie.get('path', '/')
)
return session
Session Timeout Çözümü
DMO portalı 30 dakika hareketsizlikte oturumu kapatıyordu. Çözüm: arka planda periyodik keepalive isteği.
import threading, time
class SessionKeepAlive(threading.Thread):
def __init__(self, page, interval=600): # 10 dakikada bir
super().__init__(daemon=True)
self.page = page
self.interval = interval
self._stop = threading.Event()
def run(self):
while not self._stop.wait(self.interval):
try:
# Görünmez bir element üzerine mouse move
self.page.mouse.move(100, 100)
# Hafif bir scroll
self.page.evaluate("window.scrollBy(0, 1)")
self.page.evaluate("window.scrollBy(0, -1)")
except Exception:
pass # Sayfa kapandıysa sessizce geç
def stop(self):
self._stop.set()
5900 Veriyi Paralel İşleme
Sıralı işlemde 5900 YK numarası yaklaşık 6-8 saat sürüyordu. ThreadPoolExecutor ile paralel işleme yaparak bu süreyi 90 dakikaya indirdik.
from concurrent.futures import ThreadPoolExecutor, as_completed
def process_all_orders(yk_list, session, max_workers=8):
results = []
failed = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_yk = {
executor.submit(fetch_order_detail, yk, session): yk
for yk in yk_list
}
for future in as_completed(future_to_yk):
yk = future_to_yk[future]
try:
result = future.result(timeout=30)
results.append(result)
except Exception as e:
failed.append(yk) # Başarısız olanları tekrar dene
# Başarısız olanları tek thread ile tekrar dene
for yk in failed:
try:
result = fetch_order_detail(yk, session)
results.append(result)
except Exception:
pass # Log'a yaz
return results
max_workers değerini dikkatli seçin. Portal sunucusu çok fazla eşzamanlı bağlantıda IP'nizi geçici olarak engelleyebilir. 8 ile başlayın, stabil çalışırsa artırın.
Veritabanı Karşılaştırması
Her çevrimde portaldan gelen veri SQL Server'daki mevcut veriyle karşılaştırılıyor. Sadece değişen kayıtlar güncelleniyor — bu hem performansı artırıyor hem de değişiklik logunu temiz tutuyor.
def upsert_order(conn, order: dict):
existing = conn.execute(
"SELECT durum, tahmini_sevk FROM siparisler WHERE yk_no = ?",
(order['yk_no'],)
).fetchone()
if not existing:
# Yeni kayıt
insert_order(conn, order)
log_change(conn, order['yk_no'], 'YENI', None, order['durum'])
elif existing[0] != order['durum']:
# Durum değişti
update_order(conn, order)
log_change(conn, order['yk_no'], 'DURUM_DEGISTI', existing[0], order['durum'])
send_alert(order) # Telegram/mail alarm
Tkinter Arayüzü
Thread safety Tkinter'ın en büyük tuzağı. Widget'ları arka plan thread'inden doğrudan güncellemek crash'e yol açıyor. Çözüm: root.after() ile ana thread'e mesaj göndermek.
# YANLIŞ — doğrudan yazma
def background_task():
log_widget.insert(tk.END, "Mesaj") # Crash riski!
# DOĞRU — root.after() ile
def background_task():
root.after(0, lambda: log_widget.insert(tk.END, "Mesaj\n"))