Gelen klasör deseni: paylaşımlı klasöre düşen dosyayı güvenle işleyen hat

Gelen klasör deseni, bir tedarikçinin, bankanın ya da başka bir birimin paylaşımlı klasöre bıraktığı dosyayı insan eli değmeden alıp işleyen en eski ve en dayanıklı entegrasyon biçimlerinden biridir. API'si olmayan muhasebe programı dışa aktarımını klasöre yazar, SFTP ile gelen sipariş dosyası aynı dizine düşer, ERP günde bir kez stok listesini oraya bırakır. Betiğin işi, bu klasörü düzenli aralıklarla tarayıp yeni dosyayı bulmak, doğrulamak, işlemek ve sonucunu kayda geçirmektir.

Kulağa basit gelir, ama sahada üç sorun tekrar eder: betik dosyayı yazımı bitmeden okur ve yarım veri işler; aynı dosya iki kez işlenip kayıtlar çiftlenir; hatalı dosya sessizce atlanır ve haftalar sonra fark edilir. Bu yazıda, standart kütüphaneyle bu üç sorunu çözen bir işleme hattı kuruyoruz: kararlılık kontrolü, atomik sahiplenme, içerik özetine dayalı işlem defteri ve açık bir hata klasörü.

Örnekler Python 3.8 ve üzeri sürümlerde dış bağımlılık olmadan çalışır; defter için standart kütüphanedeki sqlite3 modülü yeterlidir. Hattın dört klasörü vardır: gelen, isleniyor, tamam ve hatali. Her dosya bu klasörlerden yalnız birinde bulunur ve bulunduğu klasör o anki durumunu söyler.

Dosya tamamlandı mı: kararlılık kontrolü

Kararlılık kontrolü, bir dosyanın boyutunu ve değişiklik zamanını kısa bir aralıkla iki kez ölçüp ikisi de aynıysa yazımın bittiğine karar veren basit bir testtir. Dosya sistemi "yazma bitti" diye bir sinyal vermez; büyük bir CSV'nin paylaşımlı klasöre kopyalanması saniyeler sürer ve o sırada okuyan betik dosyanın yalnız baş kısmını görür. Yarım dosya çoğu zaman geçerli görünür, çünkü satırlar bütündür; eksik olan sondaki satırlardır.

Kontrolün yanına birkaç süzgeç eklemek gerekir. Excel açıkken oluşan ~$ ön ekli kilit dosyaları, aktarım araçlarının .part ya da nokta ile başlayan geçici adları ve sıfır baytlık dosyalar aday sayılmaz. Sıfır bayt bilinçli olarak beklemeye alınır: dosya henüz yazılmaya başlanmamış olabilir; gerçekten boşsa ileride işleme adımında yakalanır.

import time
from pathlib import Path

GELEN = Path("veri/gelen")
UZANTILAR = {".csv", ".xlsx", ".xml"}


def kararli_mi(dosya, bekleme=2.0):
    """Boyut ve değişiklik zamanı iki ölçüm arasında aynıysa dosya tamamlanmıştır."""
    ilk = dosya.stat()
    time.sleep(bekleme)
    son = dosya.stat()
    return ilk.st_size == son.st_size and ilk.st_mtime == son.st_mtime and son.st_size > 0


def adaylari_bul():
    for dosya in sorted(GELEN.iterdir(), key=lambda d: d.stat().st_mtime):
        if not dosya.is_file():
            continue
        if dosya.suffix.lower() not in UZANTILAR or dosya.name.startswith((".", "~$")):
            continue
        if kararli_mi(dosya):
            yield dosya

Daha sağlam bir yöntem, karşı tarafla anlaşmaktır: gönderen dosyayı geçici bir adla yazsın, bitince asıl adına çevirsin. Yeniden adlandırma aynı disk bölümünde atomiktir; betik ya eski adı görür ya da tamamlanmış dosyayı. Tedarikçiyle bu kuralı konuşabiliyorsanız kararlılık kontrolü ikinci bir güvence olarak kalır, konuşamıyorsanız tek güvencenizdir ve bekleme süresini dosya boyutuna göre artırmak gerekir.

Sahiplenme: dosyayı işleniyor klasörüne taşımak

Sahiplenme, betiğin işlemeye başlamadan önce dosyayı gelen klasöründen işleniyor klasörüne tek bir atomik işlemle taşımasıdır. Zamanlayıcı bir önceki çalıştırma bitmeden yenisini başlatırsa ya da iki sunucu aynı klasörü izliyorsa, aynı dosyayı iki süreç birden işleyebilir. Taşıma işlemi bunu engeller: işletim sistemi yeniden adlandırmayı ya tamamen yapar ya da hiç yapmaz, ikinci süreç dosyayı bulamaz ve sessizce vazgeçer.

import os
from pathlib import Path

ISLENIYOR = Path("veri/isleniyor")


def sahiplen(dosya):
    """Dosyayı işleniyor klasörüne taşır; aynı anda iki çalıştırma aynı dosyayı alamaz."""
    ISLENIYOR.mkdir(parents=True, exist_ok=True)
    hedef = ISLENIYOR / dosya.name
    try:
        os.rename(dosya, hedef)         # aynı disk bölümünde atomik
    except FileNotFoundError:
        return None                     # başka bir çalıştırma önce davrandı
    return hedef

Bu yaklaşımın iki koşulu vardır:

• Dört klasör aynı disk bölümünde olmalı. Bölümler farklıysa os.rename kopyala-sil olarak çalışır ve atomiklik kaybolur; ağ paylaşımlarında ise sunucunun davranışına bağlıdır, bu yüzden hattın klasörlerini yerel diske, yalnız gelen klasörünü paylaşıma koymak daha güvenlidir.
• İşleniyor klasöründe uzun süre kalan dosya bir sorunun işaretidir. Betik çökerse dosya orada kalır; başlangıçta bu klasörü tarayıp belirli bir yaştan eski dosyaları hatalıya taşımak ya da uyarı vermek gerekir.

Zamanlayıcı düzeyinde ek bir kilit, aynı betiğin iki kopyasının birlikte koşmasını baştan engeller; bunun tasarımını zamanlanmış iş tasarımı yazısında anlattım. Taşıma ile kilit birbirinin yerine geçmez; kilit süreç düzeyinde, taşıma dosya düzeyinde korur.

İşlem defteri: aynı içerik iki kez işlenmesin

İşlem defteri, işlenen her dosyanın içerik özetini, adını, boyutunu, zamanını ve sonucunu tutan küçük bir SQLite tablosudur; yeni gelen dosyanın özeti defterde "başarılı" olarak varsa dosya tekrar işlenmez. Dosya adına güvenmek yetmez: tedarikçi her hafta aynı rapor.xlsx adını kullanır, bazen de aynı dosyayı yanlışlıkla iki kez gönderir. İçeriğin SHA-256 özeti ise iki durumu da ayırt eder; ad aynı ama içerik yeni ise işlenir, ad farklı ama içerik aynı ise atlanır.

import hashlib
import sqlite3
from datetime import datetime, timezone

VERITABANI = "veri/defter.sqlite"


def defter_ac():
    baglanti = sqlite3.connect(VERITABANI)
    baglanti.execute("""
        CREATE TABLE IF NOT EXISTS islenen (
            ozet TEXT PRIMARY KEY,
            dosya_adi TEXT NOT NULL,
            boyut INTEGER NOT NULL,
            islenme_utc TEXT NOT NULL,
            sonuc TEXT NOT NULL
        )""")
    return baglanti


def dosya_ozeti(yol):
    ozet = hashlib.sha256()
    with open(yol, "rb") as f:
        for parca in iter(lambda: f.read(1024 * 1024), b""):
            ozet.update(parca)
    return ozet.hexdigest()


def daha_once_islendi_mi(baglanti, ozet):
    satir = baglanti.execute(
        "SELECT dosya_adi, islenme_utc FROM islenen WHERE ozet = ? AND sonuc = 'basarili'",
        (ozet,),
    ).fetchone()
    return satir


def deftere_yaz(baglanti, ozet, yol, sonuc):
    baglanti.execute(
        "INSERT OR REPLACE INTO islenen VALUES (?, ?, ?, ?, ?)",
        (ozet, yol.name, yol.stat().st_size, datetime.now(timezone.utc).isoformat(), sonuc),
    )
    baglanti.commit()

Defter aynı zamanda denetim izidir. "Bu ayın stok dosyası ne zaman geldi, hangi adla geldi, işlendi mi" sorusu tek bir sorguyla cevaplanır. Hatalı sonuçla kaydedilen dosya, düzeltilip yeniden gönderildiğinde içerik değiştiği için yeni bir özet alır ve normal yoldan işlenir; aynı bozuk dosya yeniden gönderilirse defter bunu da gösterir. Defteri ayrı bir dosyada tutmak yerine, varsa işlenen verinin yazıldığı veri tabanına koymak daha pratiktir; SQLite ile küçük veri boru hatları yazısındaki yapı buna uygundur.

Hattı bir araya getirmek: hatalı klasörü, log ve zamanlama

İşleme hattı, aday bulma, sahiplenme, defter sorgusu, iş mantığı ve sonuç klasörüne taşıma adımlarını tek bir döngüde birleştiren ana fonksiyondur; her adımın çıktısı bir sonraki adımın girdisidir ve her dosya döngünün sonunda tam olarak bir klasörde bulunur. İş mantığı dosyayi_isle fonksiyonunda yaşar ve hata fırlatarak konuşur: fırlatırsa dosya hatalıya, fırlatmazsa tamama gider.

import logging
import shutil
from pathlib import Path

TAMAM = Path("veri/tamam")
HATALI = Path("veri/hatali")


def dosyayi_isle(yol):
    """Kendi iş mantığınız: doğrula, dönüştür, veri tabanına yaz. Hata fırlatırsa dosya hatalıya gider."""
    if yol.stat().st_size == 0:
        raise ValueError("boş dosya")


def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    baglanti = defter_ac()
    for aday in list(adaylari_bul()):
        yol = sahiplen(aday)
        if yol is None:
            continue
        ozet = dosya_ozeti(yol)
        onceki = daha_once_islendi_mi(baglanti, ozet)
        if onceki:
            logging.info("%s atlandı: aynı içerik %s olarak %s tarihinde işlendi", yol.name, *onceki)
            TAMAM.mkdir(parents=True, exist_ok=True)
            shutil.move(str(yol), str(TAMAM / ("tekrar_" + yol.name)))
            continue
        try:
            dosyayi_isle(yol)
        except Exception:
            logging.exception("%s işlenemedi", yol.name)
            deftere_yaz(baglanti, ozet, yol, "hatali")
            HATALI.mkdir(parents=True, exist_ok=True)
            shutil.move(str(yol), str(HATALI / yol.name))
            continue
        deftere_yaz(baglanti, ozet, yol, "basarili")
        TAMAM.mkdir(parents=True, exist_ok=True)
        shutil.move(str(yol), str(TAMAM / yol.name))
        logging.info("%s tamamlandı", yol.name)


if __name__ == "__main__":
    main()

Hattı çalıştırmak için şu düzeni öneriyorum:

• Zamanlama: cron ile birkaç dakikada bir çalıştırın; dosya sistemi olaylarını dinleyen kütüphaneler ağ paylaşımlarında güvenilir çalışmaz, düzenli tarama ise her yerde çalışır.
• Hatalı klasörü boş değilse uyarı: en basit haliyle betiğin sonunda bu klasördeki dosya sayısını loga yazın ve izleme aracınız bu satırı yakalasın. Hatalı dosya sessiz kalmamalıdır.
• Tamam klasörünü budama: işlenen dosyayı belirli bir süre saklayıp sonra silin ya da arşive taşıyın. Defter kaydı kalır, dosya kalmaz.
• Doğrulamayı iş mantığının başına koyun: sütun adları, zorunlu alanlar ve tarih biçimi beklenenle uyuşmuyorsa dosya daha ilk satırda reddedilsin.

Güvenlik ve sorumluluk sınırı

Paylaşımlı klasör, dışarıdan içeriye açılan bir kapıdır ve betik o kapıdan geleni koşulsuz işlememelidir. Uzantı listesi, gelen dosyanın türünü sınırlar; betik hiçbir dosyayı çalıştırmaz, makro içeren Excel biçimlerini kabul etmez ve sıkıştırılmış dosya açıyorsa içindeki yolların hedef klasörün dışına çıkmadığını denetler. Paylaşımın yazma izni yalnız gönderen hesaba, okuma ve taşıma izni yalnız betiğin hesabına verilir.

Dosyalar kişisel veri taşıyorsa (müşteri listesi, bordro, sipariş adresleri) Kişisel Verilerin Korunması Kanunu kapsamındaki yükümlülükler hattın tasarımına girer: tamam klasörünün saklama süresi şirketin saklama politikasıyla uyumlu olmalı, log satırlarına kişisel veri yazılmamalı, defterde yalnız dosya adı ve özet tutulmalıdır. Hattın kendisi veri işleme faaliyetinin bir parçasıdır ve envanterde yer almalıdır.

Sonuç: dört klasör, bir defter

Gelen klasör deseni, dört klasör ve bir defterle yarım dosya, çift işleme ve sessiz hata sorunlarının üçünü birden çözer: kararlılık kontrolü yarım dosyayı bekletir, atomik taşıma aynı dosyayı iki sürece vermez, içerik özeti aynı veriyi iki kez işletmez, hatalı klasörü ve defter her sonucu görünür kılar. Hattın hiçbir parçası dış bağımlılık gerektirmez ve her parça tek başına test edilebilir.

İlk adım olarak, bugün elle takip ettiğiniz bir dosya akışını seçin ve yalnız kararlılık kontrolü ile sahiplenme adımlarını kurun; iş mantığını önce dosyayı doğrulayıp loga yazmakla sınırlayın. Birkaç gün sorunsuz çalıştıktan sonra defteri ve gerçek işlemeyi ekleyin.

Birden fazla kaynaktan gelen dosyaları aynı hatta bağlamak ya da mevcut akışınızı bu desene taşımak isterseniz iletişim bölümünden yazabilirsiniz.