İş uygulamalarıyla entegrasyon: Google Sheets, e-posta, Teams, Slack ve webhook

Neden önemli? Verinin yaşadığı yerler

Şirketlerde verinin önemli bir kısmı veritabanında değil, iş uygulamalarında yaşar: satış ekibinin hedef tablosu Google Sheets'te, bayilerin günlük raporu e-posta ekinde, alarmlar Teams ya da Slack kanalında, ödeme sağlayıcısının bildirimi bir webhook isteğinde. Otomasyon uzmanının işi bu adaları birbirine bağlamaktır: e-postadaki eki okuyup tabloya yazmak, tabloda bir değer değişince kanala haber vermek, dış sistemden gelen bildirime güvenip güvenmeyeceğine karar vermek.

HTTP ve REST API istemcisi: requests, httpx ve dayanıklı istekler dersinde REST istemcisi yazmayı, Yapılandırma dosyaları: JSON, YAML, TOML, INI ve ortam değişkenleri dersinde sırları ortam değişkeninde tutmayı, webhook imzasını da API tasarımına giriş: HTTP, doğrulama ve ilk servis dersinde ayrıntılı göreceksin. Burada bunları gerçek iş uygulamalarına uyguluyoruz. Tarayıcıda ağ olmadığı için gspread, imaplib ve Microsoft Graph kodlarını gri kutularda göreceksin; alıştırmalarda ise sahte istemcilerle (fake client) çalışacaksın. E-posta ayrıştırma (email modülü) ve imza doğrulama (hmac) ise standart kütüphanede olduğu için tarayıcıda gerçekten çalışır.

Entegrasyon kalıpları: çekme mi, itme mi?

Polling'in kalbi imleçtir: "en son nereye kadar okudum?" bilgisini (son updated_at ya da son kimlik) saklar ve bir sonraki sorguda yalnız ondan sonrasını istersin. Aşağıda Anadolu Market'in fiyat değişikliği tablosunu (price_updates) sayfa sayfa dönen sahte bir API'den çekiyoruz:

import veri

guncellemeler = sorted(veri.kayitlar("anadolu_market", "price_updates"), key=lambda r: r["updated_at"])


def sahte_api(since, limit=5):
    """GET /price-updates?since=...&limit=... taklidi"""
    return [r for r in guncellemeler if r["updated_at"] > since][:limit]


imlec = "2024-11-01 00:00:00"          # bir önceki çalıştırmadan kalan
yeni = []
while sayfa := sahte_api(imlec):
    yeni.extend(sayfa)
    imlec = sayfa[-1]["updated_at"]    # imleci ilerlet
print(len(yeni), "yeni güncelleme, yeni imleç:", imlec)

Google Sheets: gspread ve servis hesabı

Sheets'e betikten erişmenin en temiz yolu bir servis hesabıdır (service account): Google Cloud'da bir "robot kullanıcı" açarsın, JSON anahtarını indirirsin ve tabloyu robotun e-posta adresiyle paylaşırsın. Kişisel hesabınla OAuth yapmaktan farkı, betiğin senin oturumuna bağlı olmamasıdır.

# pip install gspread
import gspread

gc = gspread.service_account(filename="/etc/gizli/sheets-robot.json")   # repoya KOYMA
ws = gc.open_by_key("1AbC...xyz").worksheet("Bayi Satış")

satirlar = ws.get_all_values()              # her hücre METİN döner
ws.update("B5:D5", [["İzmir", "2024-01-07", 0]])
ws.append_row(["Yeni Bayi", "Bursa", "2024-01-08", 2040])
ws.batch_update([{"range": "D2:D4", "values": [[1250.5], [980], [0]]}])   # tek çağrıda çok aralık

Sheets'e yazan betik idempotent olmalıdır: iki kez çalışınca satırlar ikiye katlanmamalı. Bunun yolu "varsa güncelle, yoksa ekle" (upsert) mantığıdır. Sheets API'nin dakikalık kota sınırı olduğu için satır satır okumak yerine tabloyu bir kez okuyup bellekte anahtar → satır numarası haritası kurarsın:

class SahteSheet:
    """gspread Worksheet'in küçük bir taklidi: hücre değerleri her zaman METİN döner."""
    def __init__(self, satirlar):
        self._v = [[str(h) for h in s] for s in satirlar]
        self.cagri = 0                      # API çağrısı sayacı (kota!)

    def get_all_values(self):
        self.cagri += 1
        return [list(s) for s in self._v]

    def update_row(self, satir_no, degerler):          # 1 tabanlı satır no
        self.cagri += 1
        self._v[satir_no - 1] = [str(d) for d in degerler]

    def append_row(self, degerler):
        self.cagri += 1
        self._v.append([str(d) for d in degerler])


ws = SahteSheet([["Bayi", "Şehir", "Tutar"], ["Merkez Bayi", "İstanbul", 1250.5], ["Kızılay Bayi", "Ankara", 980]])


def upsert(ws, satir):
    tum = ws.get_all_values()
    konum = {s[0]: i + 1 for i, s in enumerate(tum)}     # anahtar -> 1 tabanlı satır no
    if satir[0] in konum:
        if tum[konum[satir[0]] - 1] != [str(d) for d in satir]:
            ws.update_row(konum[satir[0]], satir)
    else:
        ws.append_row(satir)


for _ in range(2):                                        # iki kez çalıştır
    upsert(ws, ["Kızılay Bayi", "Ankara", 2410])
    upsert(ws, ["Nilüfer Bayi", "Bursa", 2040])
print(ws.get_all_values())
print("API çağrısı:", ws.cagri)

E-posta okuma: imaplib ve email modülü

Bayiler, bankalar ve tedarikçiler raporlarını hâlâ çoğunlukla e-posta ekiyle gönderir. IMAP ile posta kutusuna bağlanır, ölçüte göre arar (UNSEEN FROM "..." SINCE 01-Oct-2024), iletiyi ham bayt olarak indirirsin. Gerisini, yani başlıkları, gövdeyi ve ekleri ayrıştırmayı standart kütüphanenin email modülü yapar.

import imaplib, os
from email import message_from_bytes, policy

with imaplib.IMAP4_SSL("imap.gmail.com") as imap:
    imap.login(os.environ["MAIL_KULLANICI"], os.environ["MAIL_UYGULAMA_SIFRESI"])
    imap.select("INBOX")
    durum, veri = imap.search(None, '(UNSEEN FROM "kadikoy@ilgazbank.example" SINCE 01-Oct-2024)')
    for num in veri[0].split():
        durum, parca = imap.fetch(num, "(RFC822)")
        ileti = message_from_bytes(parca[0][1], policy=policy.default)
        for ek in ileti.iter_attachments():
            if ek.get_filename().endswith(".csv"):
                isle(ek.get_content())                     # metin ya da bayt
        imap.store(num, "+FLAGS", "\\Seen")               # işlendi olarak işaretle

Aşağıdaki kod Ilgaz Bank'ın 15 Mart 2024 işlemlerinden ekli bir CSV içeren gerçek bir e-posta üretiyor, bayta çeviriyor (IMAP'ten böyle gelir) ve geri ayrıştırıyor:

import csv
import io
import veri
from email.message import EmailMessage
from email import message_from_bytes, policy

txns = veri.kayitlar("ilgaz_bank", "bank_txns")


def rapor_epostasi(gun):
    tampon = io.StringIO()
    yazici = csv.writer(tampon)
    yazici.writerow(["txn_id", "amount", "channel"])
    for t in txns:
        if t["txn_time"].startswith(gun):
            yazici.writerow([t["txn_id"], t["amount"], t["channel"]])
    m = EmailMessage()
    m["From"] = "Kadıköy Şubesi <kadikoy@ilgazbank.example>"
    m["To"] = "operasyon@ilgazbank.example"
    m["Subject"] = f"Günlük işlem raporu {gun}"
    m.set_content("Merhaba,\nGünlük işlem dökümü ektedir.")
    m.add_attachment(tampon.getvalue().encode("utf-8"), maintype="text", subtype="csv",
                     filename=f"islemler_{gun}.csv")
    return m.as_bytes()


ham = rapor_epostasi("2024-03-15")
ileti = message_from_bytes(ham, policy=policy.default)
print("Konu  :", ileti["subject"])
print("Kimden:", ileti["from"].addresses[0].display_name)
print("Gövde :", ileti.get_body(("plain",)).get_content().splitlines()[1])
for ek in ileti.iter_attachments():
    satirlar = list(csv.DictReader(io.StringIO(ek.get_content())))
    print("Ek    :", ek.get_filename(), len(satirlar), "satır")

Microsoft Graph, Teams ve Slack

Microsoft 365 kullanan şirketlerde Outlook postası, Teams ve OneDrive'a Microsoft Graph API'siyle erişilir. Kimlik doğrulama Azure'da kayıtlı bir uygulama ve MSAL kütüphanesiyle yapılır. Yalnız bir kanala mesaj atmak istiyorsan daha basit yol gelen webhook'tur (Teams'te Workflows/Power Automate ile, Slack'te "Incoming Webhook" uygulamasıyla oluşturulur): sana gizli bir URL verilir, o adrese JSON göndermek yeter.

# pip install msal requests
import msal, requests, os
app = msal.ConfidentialClientApplication(os.environ["AZ_CLIENT_ID"],
        authority=f"https://login.microsoftonline.com/{os.environ['AZ_TENANT']}",
        client_credential=os.environ["AZ_SECRET"])
token = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])["access_token"]
r = requests.get("https://graph.microsoft.com/v1.0/users/rapor@lezzetkurye.example/messages"
                 "?$filter=hasAttachments eq true&$top=10",
                 headers={"Authorization": f"Bearer {token}"}, timeout=10)

# Slack ve Teams gelen webhook: URL bir sırdır, ortam değişkeninde dursun
requests.post(os.environ["SLACK_WEBHOOK_URL"], json=yuk, timeout=10).raise_for_status()
requests.post(os.environ["TEAMS_WEBHOOK_URL"], json=kart, timeout=10).raise_for_status()

Webhook'un istediği tek şey doğru biçimde bir JSON yüküdür (payload). Slack'te Block Kit, Teams'te Adaptive Card kullanılır. Yükü bir fonksiyonla üretmek onu test edilebilir yapar. Lezzet Kurye'nin 31 Aralık 2024 sipariş özetini Slack bloğuna çeviriyoruz:

import json
from collections import Counter
import veri

siparisler = veri.kayitlar("lezzet_kurye", "lk_orders")
gun = [s for s in siparisler if s["placed_at"].startswith("2024-12-31")]
durum = Counter(s["status"] for s in gun)

yuk = {
    "text": f"31.12.2024 özeti: {len(gun)} sipariş",          # bildirim önizlemesi
    "blocks": [
        {"type": "header", "text": {"type": "plain_text", "text": "Lezzet Kurye günlük özet"}},
        {"type": "section", "fields": [
            {"type": "mrkdwn", "text": f"*Teslim*\n{durum['teslim']}"},
            {"type": "mrkdwn", "text": f"*İptal*\n{durum['iptal']}"},
            {"type": "mrkdwn", "text": f"*İade*\n{durum['iade']}"},
        ]},
    ],
}
print(json.dumps(yuk, ensure_ascii=False, indent=1)[:260])

Gelen webhook'u doğrulamak: HMAC imzası

Webhook adresin internete açıktır; herkes oraya "ödeme alındı" diye istek atabilir. Bu yüzden sağlayıcılar (Stripe, GitHub, Slack, iyzico) isteği ortak bir gizli anahtarla HMAC-SHA256 ile imzalar ve imzayı bir başlıkta gönderir. Alıcı aynı imzayı ham gövde baytları üzerinden yeniden hesaplar ve hmac.compare_digest ile karşılaştırır. Slack ve Stripe imzaya zaman damgasını da katar; 5 dakikadan eski istekler reddedilerek tekrar saldırısı (replay) önlenir.

import hashlib
import hmac
import json

GIZLI = b"whsec_lezzet_test"              # gerçekte ortam değişkeninden
govde = b'{"event_id": "ev_101", "type": "siparis.teslim", "order_id": 2}'
imza = "sha256=" + hmac.new(GIZLI, govde, hashlib.sha256).hexdigest()


def dogrula(govde, imza_basligi, gizli):
    beklenen = "sha256=" + hmac.new(gizli, govde, hashlib.sha256).hexdigest()
    return hmac.compare_digest(beklenen, imza_basligi)


print(dogrula(govde, imza, GIZLI))
print(dogrula(govde.replace(b"2}", b"9}"), imza, GIZLI))        # gövde değiştirildi
print(dogrula(json.dumps(json.loads(govde)).encode(), imza, GIZLI))

Üçüncü satır neden False? Gövdeyi JSON olarak ayrıştırıp yeniden yazmak boşlukları ve anahtar sırasını değiştirebilir; imza bayt bayt aynı metin üzerinden hesaplanmalıdır. FastAPI'de ham gövdeyi await request.body() ile alırsın:

from fastapi import FastAPI, Request, Header, HTTPException
app = FastAPI()

@app.post("/webhook/odeme")
async def odeme(request: Request, x_imza: str = Header(...)):
    govde = await request.body()                      # ham bayt, json() DEĞİL
    if not dogrula(govde, x_imza, GIZLI):
        raise HTTPException(401, "imza geçersiz")
    olay = json.loads(govde)
    if zaten_islendi(olay["event_id"]):               # sağlayıcı tekrar gönderebilir
        return {"durum": "tekrar"}
    kuyruga_ekle(olay)                                # ağır işi arka plana ver
    return {"durum": "alındı"}                        # hızlı 2xx dön, yoksa yeniden dener

Sırlar, n8n ve Make

Servis hesabı JSON'u, IMAP uygulama şifresi, webhook URL'leri ve imza anahtarları sırdır: koda, Git'e ve log'a girmez; ortam değişkeni ya da kasa (Azure Key Vault, AWS Secrets Manager) ile verilir, düzenli döndürülür (rotate) ve her entegrasyon için ayrı, en az yetkili anahtar kullanılır. OAuth'ta erişim belirteci (access token) kısa ömürlüdür; yenileme belirteci (refresh token) ise şifre kadar değerlidir.

n8n ve Make gibi düşük kodlu araçlar tetikleyici, bağlayıcı ve zamanlama işini hazır sunar: "Gmail'e ek gelince Sheets'e yaz" akışı dakikalar içinde kurulur. Python'un rolü, bu akışın ortasındaki zor adımdır: karmaşık ayrıştırma, doğrulama, hesaplama, model çağrısı. n8n'de "Code" düğümü ya da senin yazdığın küçük bir HTTP servisi (webhook alıcısı) bu adımı üstlenir. İş bölümü kuralı: bağlantı ve tetikleme platformda, iş kuralı ve test edilebilir mantık Python'da.