ای ٹی ایل پائپ لائنز ہینڈ بک: ازگر میں پروڈکشن گریڈ پائپ لائنز کیسے بنائیں

سیلاب کے خطرے سے باخبر رہنے کے لیے کچھ غیر واضح لیکن ضروری ہے: صاف، منظم ڈیٹا۔

اس ٹیوٹوریل میں، آپ اپنا ڈیٹا پائپ لائن بناتے ہیں۔ فرانس کے آفیشل اوپن واٹر ڈیٹا API Hub’Eau سے روزانہ پانی کی سطح کی ریڈنگ حاصل کرنے کے لیے Python ETL (ایکسٹریکٹ، ٹرانسفارم، لوڈ) پائپ لائن بنائیں۔ پھر ہم اس ڈیٹا کو صاف کرتے ہیں اور اسے لائیو ورژن کی طرح ایک عوامی ڈیٹاسیٹ کے طور پر شائع کرتے ہیں۔

یہ ٹیوٹوریل ایک حقیقی پائپ لائن پر مبنی ہے جو ہفتے میں ایک بار شیڈول کے مطابق چلتی ہے اور پیرس سیلاب کے ڈیٹاسیٹ کو خود بخود اپ ڈیٹ کرتی ہے۔

لیکن آپ صرف کوڈ کو کاپی اور پیسٹ نہیں کرتے ہیں۔ اصل مقصد سمجھنا ہے۔ کیوں پائپ لائن توقع کے مطابق کام کرتی ہے۔ آئیے اسکرپٹس کو الگ کرنے کے ڈیزائن کے فیصلے پر نظر ڈالیں جس کا مقصد اسکرپٹس سے ایک بار چلنا ہے جو سالوں تک بغیر توجہ کے چلتی رہیں گی۔

آپ اس لیپ ٹاپ سے کوڈ کر سکتے ہیں۔ زیادہ تر سبق کے لیے، پائپ لائن اس پر چلتی ہے: نقلی (مذاق) API ڈیٹا. یہ آپ کو اصل سرور کے ساتھ گڑبڑ کیے بغیر کسی بھی سیل کو محفوظ طریقے سے چلانے کی اجازت دیتا ہے۔ بعد کا سیکشن آپ کو دکھاتا ہے کہ لائیو API میں کیسے منتقل کیا جائے۔

آخر میں، آپ اس قابل ہو جائیں گے:

• نکالنے، تبدیلی، اور بوجھ کے نمونوں کی وضاحت اور نفاذ کریں۔

• ازگر کے ساتھ کنفیگریشن کا انتظام @dataclass جگہ جگہ مستقل بکھرنے کے بجائے

• API بازیافت کوڈ لکھیں جو نیٹ ورک کی ناکامیوں اور صفحہ بندی کے جوابات سے بچ جاتا ہے۔

• کسی ایک غلط قطار کو پوری پائپ لائن پر عمل درآمد کو روکنے سے روکنے کے لیے سخت قسم کا جبر نافذ کرتا ہے۔

• انکریمنٹل ڈیٹا کو محفوظ طریقے سے ڈپلیکیٹ اور انضمام کریں۔

• ہر چیز کو ایک قابل تقلید، شیڈول کے مطابق ہستی سے جوڑنا main() داخلہ پوائنٹ

اشاریہ:

شرطیں

• Python 3.10+. کوڈ قسم کے اشارے اور ڈیٹا کلاسز کا استعمال کرتا ہے۔ یہ Python 3.7 سے شروع ہوتا ہے، لیکن 3.10+ بہترین ہے۔

• پانڈا ڈیٹا فریم کا علم: ڈیٹا پڑھیں، قطاروں کو فلٹر کریں، اور کالم کے بنیادی کام انجام دیں۔

• آرام افعال اور بنیادی OOP (آبجیکٹ اورینٹڈ پروگرامنگ) ازگر۔ فکر نہ کرو۔ یہ گائیڈ کسی بھی ایسی چیز کی وضاحت کرے گا جو آپ کے جاتے وقت واضح نہ ہو۔

• اختیاری: ایک مفت Kaggle اکاؤنٹ اور Kaggle CLI (صرف اس صورت میں جب آپ حتمی اشاعت کے مراحل کو چلانا چاہتے ہیں)

انحصار انسٹال کریں۔

pip install requests pandas numpy ipykernel

آپ Jupyter نوٹ بک کے اندر کام کریں گے۔ ڈاؤن لوڈ کریں .ipynb آپ Kaggle سے فائل ڈاؤن لوڈ کر سکتے ہیں یا Kaggle پر براہ راست کام کرنے کے لیے "کاپی اور ترمیم کریں” پر کلک کر سکتے ہیں۔ اس کے لیے آپ کو ایک Kaggle اکاؤنٹ کی ضرورت ہوگی۔

اختیاری: اگر آپ مقامی طور پر نوٹ بک چلانا چاہتے ہیں۔ notebook پیکیج بھی:

pip install notebook

حصہ 1: ڈیزائن منطق

کوڈ کی ایک لائن کو چھونے سے پہلے ہم 3 منٹ گزارتے ہیں۔ کیوں پائپ لائن اس طرح نظر آتی ہے۔

یہ ہے بڑی تصویر دیکھیں. کوڈ کی سطح کا ہر فیصلہ بعد میں ان خیالات میں سے کسی ایک کا پتہ لگاتا ہے۔ اس حصے کو پڑھیں یہاں تک کہ اگر آپ باقی سب کچھ چھوڑ دیں۔

ETL پائپ لائن کیا ہے؟

ETL کا مطلب ہے۔ نکالنا، تبدیل کرنا، لوڈ کرنا. یہ ڈیٹا کو ماخذ سے منزل تک قابل اعتماد اور دوبارہ قابل اعتماد انداز میں منتقل کرنے کا ایک معیاری نمونہ ہے۔

• نکالنا: APIs، ڈیٹا بیس اور فائلوں جیسے ذرائع سے ڈیٹا درآمد کریں۔

• تبدیلی: اپنے ڈیٹا کو صاف، معیاری، افزودہ اور درست کریں۔

• لوڈ: نتائج کو کسی منزل پر لاگ ان کریں جیسے گودام، CSV، یا عوامی پلیٹ فارم۔

یہاں یہ ہے کہ یہ تین مراحل اس منصوبے کا نقشہ کیسے بناتے ہیں:

قدم	یہاں کیا ہوتا ہے
نکالنا	اگر آپ کے پاس موجودہ ڈیٹا ہے تو اسے لوڈ کریں اور پھر (بذریعہ) Hub’Eau API کو کال کریں۔ requests) ہر پیمائشی اسٹیشن کے لیے
تبدیلی	فرانسیسی کالموں اور آئٹمز کا انگریزی میں ترجمہ کریں، ڈیٹا کی قسمیں درست کریں، اور نقلیں ہٹا دیں۔
لوڈ	CSV اور میٹا ڈیٹا فائلیں بنائیں اور انہیں CLI کے ذریعے Kaggle پر شائع کریں۔

فن تعمیر کا جائزہ

"اقتباس” پہلے ہی دو مختلف ذرائع سے منسلک ہے۔ موجودہ ڈیٹاسیٹ (جو آپ کے پاس پہلے سے ہے) نیا API سے ڈیٹا۔ یہ امتیاز اگلے خیال کا بیج ہے۔

اسے پروڈکشن گریڈ بنانے کے لیے دو پیٹرن

دو نمونے ہیں جو اس پائپ لائن کو محفوظ طریقے سے چلانا ممکن بناتے ہیں اور سالوں تک غیر توجہ شدہ شیڈولنگ کے ساتھ۔

1. بے حسی۔

بے حسی اس کا مطلب ہے کہ ایک ہی کام کو دو بار چلانے سے وہی نتائج برآمد ہوں گے جیسے ایک بار چلائے گئے تھے۔ نقل ہی وہ چیز ہے جو اس پائپ لائن کو کمزور بناتی ہے۔ اگر شیڈیولر غلطی سے دو بار ٹرگر ہو جاتا ہے یا نیٹ ورک کی دوبارہ کوشش اسی دن اسے دوبارہ لاتی ہے، تو دوسری رن ڈپلیکیٹ قطاریں نہیں بنائے گی۔ یہ کسی بھی چیز کے لئے بہت اہم ہے جو کسی کے دیکھے بغیر شیڈول پر چلتا ہے۔

2. اضافی لوڈنگ

بولی پائپ لائن ہے پوری تمام رنز کا ریکارڈ۔ یہ سست ہے، API کوٹہ ضائع کرتا ہے، اور کمزور ہے۔ آپ جتنا زیادہ ڈیٹا منتقل کریں گے، اس کے ناکام ہونے کا امکان اتنا ہی زیادہ ہے۔

نہیں اضافہ پائپ لائنیں اس کو روکتی ہیں۔ اس کے بجائے، یہ ہے:

• ڈیٹا سیٹ میں پہلے سے ہی تازہ ترین تاریخ کو چیک کریں۔

• صرف ڈیٹا کی درخواست کریں۔ اس تاریخ سے

• نئے ریکارڈز کو موجودہ ڈیٹاسیٹ میں ضم کریں۔

یہ منطق واضح طور پر بنائی گئی ہے۔ determine_update_range.

ہمیشہ اپنے آپ سے پوچھیں: "کیا مجھے واقعی یہ کرنا ہے؟” ان عادات میں سے ایک کمزور اسکرپٹ کو پروڈکشن پائپ لائن سے الگ کرنا ہے۔ مثال کے طور پر، اس پائپ لائن کے تمام "مہنگے” یا "بیرونی” مراحل (نیٹ ورک کالز، ڈسک رائٹ، پوسٹس) سب سے پہلے سستے مقامی چیک کے ذریعے محفوظ ہوتے ہیں۔

حصہ 2: ترتیب اور انحصار

یہاں امپورٹ بلاک ہے۔ یہ غیر واضح معلوم ہو سکتا ہے، لیکن جس طرح سے اسے منظم کیا گیا ہے وہ قابل ذکر ہے۔

# Standard library imports
import json
import os
import random
import subprocess
from dataclasses import dataclass, field
from datetime import date, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple

# Third-party libraries
import numpy as np
import pandas as pd
import requests

# Display config for notebooks
pd.set_option('display.max_columns', None)   # All columns will show
pd.set_option('display.max_colwidth', None)  # Prevents cutting long column text with ...

کوڈ کی سطح کے نوٹس:

• درآمدات کو تین بلاکس میں تقسیم کیا گیا ہے: معیاری لائبریری، تیسری پارٹی، مقامیدرمیان میں ایک خالی لکیر ہے۔ یہ پی ای پی 8 سے درآمدی آرڈر کے قوانین کی پیروی کرتا ہے۔ زیادہ تر خودکار فارمیٹرز (isort، ruff) اسی گروپ بندی کو لاگو کرتا ہے۔

• اگلا، کچھ بھی درآمد نہیں کیا گیا تھا. from module import *. یہ بیان موجودہ نام کی جگہ کو آلودہ کرتا ہے۔ جب کوئی آپ کے کوڈ کو چھ ماہ بعد ڈیبگ کرتا ہے، تو اس کا پتہ لگانا مشکل ہوتا ہے کہ نام کہاں سے آیا ہے۔ Python سٹائل گائیڈز اس مشورے کی بازگشت کرتے ہیں، بشمول Google کی Python سٹائل گائیڈ۔

• کہ pd.set_option(...) کال خالصتاً وجودی ہے۔ لیپ ٹاپ پڑھنے کی اہلیتاتنے وسیع ڈیٹا فریموں کو چھوٹا نہیں کیا جائے گا۔ .... پائپ لائن کی منطق پر اس کا کوئی اثر نہیں ہوتا۔ عام طور پر آپ اسے کسی اور طریقے سے ہٹا دیں گے یا دائرہ کار بنائیں گے۔ .py سکرپٹ

حصہ 3: ڈیٹا کلاسز کا استعمال کرتے ہوئے ترتیب کو منظم کرنا

آپ کو کمپوزیشن لیئرز کا خیال کیوں رکھنا چاہیے؟

ہر پائپ لائن میں دستک: نگرانی کے لیے اسٹیشن، سیلاب کی حد اور پوسٹنگ کے مقامات۔ ایک برا طریقہ یہ ہے کہ جب آپ اسے لکھتے ہیں تو ان اقدار کو اپنے پورے کوڈ میں لفظی طور پر چھڑکنا ہے۔ تم ختم کرو if level > 6000: تین افعال گہرائی میں دب گئے۔ پھر تبدیل کریں کوئی بھی اسے ترتیب دینے کا مطلب پوری فائل کو اسکین کرنا ہے، اور ایک پوائنٹ کو اپ ڈیٹ کرنا اور دوسرے کو یاد کرنا آسان ہے۔

اصلاحات: اپنی تمام ترتیبات کو ایک جگہ پر سنٹرلائز کریں۔ ازگر کی @dataclass ڈیکوریٹر اس کے لیے بہترین ہیں۔

خصوصیت	یہ کیوں ضروری ہے؟
خود کار طریقے سے پیدا __init__، __repr__، __eq__	آپ کو خود لکھنے کی ضرورت نہیں ہے۔
اشارہ درج کریں۔	IDE خودکار تکمیل اور خود دستاویزی کوڈ فراہم کرتا ہے۔
اختیاری frozen=True	اگر آپ کنفیگریشن نوبس چاہتے ہیں جنہیں تخلیق کے بعد تبدیل نہیں کیا جا سکتا ہے تو یہ حقیقی تبدیلی فراہم کرتا ہے۔
__post_init__ ہک	تخلیق کے فوراً بعد ایک بار اخذ کردہ فیلڈز کی توثیق یا حساب لگائیں۔

اس کا موازنہ عام طور پر لکھی گئی ترتیب سے کریں۔ dict:

config = {
    "stations": ["STN001", "STN002"],
    "flood_threshold": 6000,
    "publish_url": "https://example.com/alerts",
    "retry_count": 2,
    "timeout_seconds": 5,
}

باقاعدگی سے لغات میں تبدیلی، ٹائپ چیکنگ، اور خودکار تکمیل کی خصوصیات فراہم نہیں ہوتی ہیں۔

کوڈ لیول واک تھرو

پائپ لائن تین کنفیگریشن کلاسز کی وضاحت کرتی ہے: ہر شخص کے پاس ہوتا ہے۔ ایک ذمہ داری: API تفصیلات، اسٹیشن کے قواعد اور اشاعت کے مقامات۔ ہر طبقہ ایک ہو جاتا ہے۔ ماڈیول لیول سنگلٹن مثال. پائپ لائن میں دیگر تمام فنکشنز ان سنگل ٹن سے پڑھتے ہیں۔

@dataclass
class APIConfig:
    """API configuration for HubEau data fetching.

    Think of this as the "address book" for the API.
    """
    use_mock: bool = True
    base_url: str = "https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab"
    metric: str = "HIXnJ"  # Daily max water level (elaborated observations)
    # Pagination: fetch 20k records per request (API Limit)
    max_per_page: int = 20000
    timeout_seconds: int = 60  # Network timeout

    def __post_init__(self):
        """Validate configuration after initialization."""
        if self.max_per_page <= 0:
            raise ValueError("max_per_page must be positive")
        if self.timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be positive")

اندرونی توثیق کی جانچ کریں۔ __post_init__. یہ خود بخود بننے کے فوراً بعد عمل میں لایا جاتا ہے۔ __init__. ناقص تعمیر APIConfig(max_per_page=-1) اصل پائپ لائن کے عمل کے دوران بعد میں بگ کے طور پر ظاہر ہونے کے بجائے، یہ شروع ہونے پر زور سے اور فوری طور پر ناکام ہو جاتا ہے۔

@dataclass
class StationConfig:
    """Station monitoring configuration.

    The 'what' of data collection: which stations, what's a flood?
    """
    station_codes: List[str] = field(default_factory=lambda: [
        "F700000109", "F700000110", "F700000111",
        "F700000102", "F700000103",
    ])
    flood_threshold_mm: int = 6000  # Flood alert threshold
    earliest_date: str = "1900-01-01"  # How far back to go

قابل تبدیلی ڈیفالٹ ٹریپس: قریب سے دیکھیں station_codes. کے طور پر نہیں لکھا ہے۔ station_codes: List[str] = [...]. یہ جان بوجھ کر ہے اور ازگر کے سب سے عام مسائل میں سے ایک سے بچتا ہے۔ ڈیفالٹ آرگیومینٹس یا ڈیٹا کلاس فیلڈز کے بطور باقاعدہ تبدیل ہونے والی اشیاء (فہرستیں، لغات، یا سیٹ) استعمال کرتے وقت، تمام مثالیں ایک ہی بنیادی چیز کا اشتراک کرتی ہیں۔. اگر آپ اسے ایک مثال میں تبدیل کرتے ہیں، تو یہ خود بخود ہر جگہ بدل جائے گا۔

اسٹیک اوور فلو اس کے بارے میں "کم سے کم حیران کن" متغیر ڈیفالٹ دلائل کے بارے میں ان کی بحث میں تفصیل سے جاتا ہے، جیسا کہ اصلی ازگر میں اختیاری دلائل کے لئے ان کی رہنمائی کرتا ہے۔ یہاں اصلاحات ہیں: field(default_factory=...). کہ تازہ فیکٹری فنکشن (یہاں: lambda) ہر نئی مثال کے لیے، ہر مثال کو اپنی آزاد فہرست ملتی ہے۔

وضاحت:

# Bad: shared mutable default
@dataclass
class BadConfig:
    station_codes: list[str] = []

a = BadConfig()
b = BadConfig()

a.station_codes.append("ALERT")
print(a.station_codes)  # ['ALERT']
print(b.station_codes)  # ['ALERT']  <-- same list!!


# Good: fresh list per instance
@dataclass
class GoodConfig:
    station_codes: list[str] = field(default_factory=list)

x = GoodConfig()
y = GoodConfig()

x.station_codes.append("ALERT")
print(x.station_codes)  # ['ALERT']
print(y.station_codes)  # []  <-- independent list

اور فائنل کنفیگ سنگلٹن، کاگل کنفیگریشن، مندرجہ ذیل ہے:

@dataclass
class KaggleConfig:
    """Kaggle dataset publishing configuration."""
    dataset_slug: str = "grimespoint/paris-flood-dataset"
    input_csv: str = "kaggle/input/datasets/{slug}/paris_flood_dataset.csv"
    output_dir: Path = field(default_factory=lambda: Path("kaggle/working/kaggle_dataset"))
    mock_output_dir: Path = field(default_factory=lambda: Path("mock_output"))
    output_filename: str = "paris_flood_dataset.csv"
    mock_output_filename: str = "mock_flood_dataset.csv"
    metadata_filename: str = "dataset-metadata.json"

    # Metadata
    title: str = "Paris flood dataset"
    keywords: list = field(default_factory=lambda: [
        "tabular", "weather and climate", "environment", "europe", "time series analysis"
    ])
    geospatial_coverage: str = "Paris, France"
    update_frequency: str = "Weekly"
    license_name: str = "CC0-1.0"

    # Computed fields (set in __post_init__)
    output_csv_path: Path = field(init=False)
    metadata_path: Path = field(init=False)

    def __post_init__(self):
        """Compute derived paths after initialization."""
        self.input_csv = self.input_csv.format(slug=self.dataset_slug)
        self.output_csv_path = self.output_dir / self.output_filename
        self.metadata_path = self.output_dir / self.metadata_filename
        self.mock_output_filename = self.mock_output_dir / self.mock_output_filename

# Initialize configs - module-level singletons
API_CONFIG = APIConfig()
STATION_CONFIG = StationConfig()
KAGGLE_CONFIG = KaggleConfig()

یہ سب سے زیادہ استعمال کیا جاتا ہے __post_init__. output_csv_path اور metadata_path یہ نشان زد ہے۔ field(init=False)، لہذا اسے کنسٹرکٹر کے ذریعے براہ راست سیٹ نہیں کیا جاسکتا۔ اس کے بجائے، __post_init__ دوسرے فیلڈ میں اس کا حساب لگائیں (output_dir اور output_filename)۔

اس پیٹرن کا استعمال کریں اخذ کردہ، حسابی قدر: دوبارہ حساب کرنے کے بجائے، آپ صرف ایک جگہ ایک بار حساب لگاتے ہیں۔ output_dir / output_filename جب بھی آپ کو اپنے کوڈ بیس میں کہیں اور راستے کی ضرورت ہو۔

اصلی پائتھون کی ڈیٹا کلاسز گائیڈ یا O'Reilly کی ڈیٹا کلاسز کا باب دیکھیں۔ Python میں روانی اس پیٹرن کے بارے میں مزید جانیں۔

ٹپ: خود کار طریقے سے پیدا __repr__ ہم مفت پڑھنے کا مواد فراہم کرتے ہیں۔ کال print(API_CONFIG): فارمیٹنگ کوڈ کی ایک لائن کے بغیر تمام فیلڈز اور اقدار دکھاتا ہے۔ پائپ لائن پر عمل درآمد کو ڈیبگ کرتے وقت فوری سنٹی چیکس کے لیے مفید ہے۔

print(APIConfig)
# prints APIConfig(use_mock=True, base_url="https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab", ...)

نکالنے کا مرحلہ سورس سسٹم سے ڈیٹا لیتا ہے اور اسے میموری میں پڑھتا ہے۔ یہاں، آپ کا مخلص دو نکالنے کا ذریعہ: موجودہ ڈیٹاسیٹ (جسے میں نے پچھلی بار پوسٹ کیا تھا) اور نیا (حالیہ) Hub'Eau API سے ڈیٹا۔

سیملیس فائل لوڈنگ

ڈیزائن منطق

جب یہ پائپ لائن پہلی بار چلائی جائے تو کیا ہونا چاہیے (اگر ابھی تک کوئی موجودہ ڈیٹا سیٹ نہیں ہے)؟ بولی نفاذ ہے FileNotFoundError.

یہ راز ہے: load_csv() درج ذیل پر عمل کریں۔ null آبجیکٹ پیٹرن. غلطی پھینکنے کے بجائے مفت ڈیٹا فریم۔ تمام ڈاون اسٹریم فنکشنز پھر "موجودہ ڈیٹا نہیں" اور "کچھ موجودہ ڈیٹا" کو اسی طرح سنبھال سکتے ہیں، بغیر کسی خاص کیس کی ضرورت ہے۔

کوڈ کی سطح کی مشقیں۔

یہ فنکشن CSV فائل کو ڈیٹا فریم میں لوڈ کرتا ہے۔ اگر فائل موجود نہیں ہے، تو یہ کریش ہونے کے بجائے ایک خالی ڈیٹا فریم لوٹاتا ہے۔

low_memory=False پانڈوں کو فائل کو غور سے پڑھنے کے لیے کہہ کر مخلوط قسم کے اندازوں کو روکتا ہے۔ parse_dates=True تاریخ جیسے کالموں کو خود بخود تاریخوں میں تبدیل کرنے کی کوشش۔ delimiter="," یہ پانڈوں کو بتاتا ہے کہ فائلوں کو کوما سے الگ کیا گیا ہے۔

def load_csv(path: str) -> pd.DataFrame:
    """Load CSV file or return empty DataFrame if file does not exist.

    Args:
        path (str): Full path to the CSV file.

    Returns:
        pd.DataFrame: Loaded data, or empty DataFrame if file not found.

    Raises:
        pd.errors.ParserError: If the CSV is malformed.
    """
    if os.path.exists(path):
        return pd.read_csv(path, low_memory=False, parse_dates=True, delimiter=",")
    return pd.DataFrame()   # Null Object: consistent return type

جانچنے کے لیے، غیر موجود راستے کے خلاف کوشش کریں۔

df_missing = load_csv("/tmp/does_not_exist.csv")
print(df_missing.empty)  # True: no crash

# Callers can always do this, instead of an `is None` check:
if df_missing.empty:
    print("No existing data. Will run a full fetch from earliest date.")

بہترین طرز عمل: واپسی مسلسل قسم فنکشن کے تمام کوڈ راستوں میں۔ ایک فنکشن جو کبھی کبھی واپس آتا ہے: DataFrame اور کبھی کبھی None تمام کال کرنے والے None ان کو استعمال کرنے سے پہلے نتائج چیک کریں۔

واپسی DataFrameچیزیں اس بات پر آسان ہو جاتی ہیں کہ آیا یہ خالی ہے یا نہیں۔ آپ کو کبھی پوچھنے کی ضرورت نہیں ہے۔ "کیا مجھے حقیقی نتائج ملے یا؟ None؟" استعمال سے پہلے۔

اضافی اپ ڈیٹ منطق

ڈیزائن منطق

یہ حصہ 1 کا "انکریمنٹل لوڈنگ" پیٹرن ہے۔ کچھ بھی لانے سے پہلے، یہ سوالات پوچھیں: "میرے پاس پہلے سے حالیہ ریکارڈ کون سے ہیں؟ کیا مجھے واقعی مزید کی ضرورت ہے؟"

حکمت عملی: چیک کریں کہ آیا موجودہ ڈیٹا پہلے سے شامل ہے۔ کل.

• مثال: اپ ڈیٹ کو مکمل طور پر چھوڑ دیں۔ کرنے کو کچھ نہیں ہے۔ ڈیٹا سیٹ اپ ٹو ڈیٹ ہے۔

• نہیں: آخری معلوم تاریخ کے ایک دن بعد بازیافت شروع ہوتی ہے۔

Existing data: Jan. 1 – Jan. 15
Yesterday: Jan. 19

Decision: fetch from Jan. 16 onwards (not from Jan. 1)

کیوں کل اور پھر نہیں۔ آج? آج کی پیمائش ابھی تک سورس سسٹم پر مکمل نہیں ہوئی ہو گی۔ Hub'Eau کے "نفیس مشاہدات" پر روزانہ مجموعی طور پر کارروائی کی جاتی ہے، لہذا سب سے محفوظ جانچ آخری کے خلاف ہے۔ مکمل طور پر مکمل دوپہر

کوڈ لیول واک تھرو

determine_update_range() تازہ ترین محفوظ شدہ تاریخ کو چیک کریں۔ آپ کو ایک پیغام نظر آئے گا کہ "یہ پہلے سے ہی اپ ٹو ڈیٹ ہے" یا "مندرجہ ذیل تاریخ سے ڈاؤن لوڈ کرنا شروع کریں۔"

مرحلہ وار:

1. فنکشن موجودہ ڈیٹا سے شروع ہوتا ہے۔

   • existing یہ ایک پانڈا ہے۔ DataFrame ڈیٹا کی کچھ قطاریں پہلے ہی موجود ہیں۔

2. اگر کوئی ڈیٹا نہیں ہے۔

3. تاریخ کا کالم تلاش کریں۔

   • کوڈ چیک کرتا ہے کہ کون سا کالم تاریخوں پر مشتمل ہے۔

   • اگر آپ کے پاس دونوں نہیں ہیں تو آپ کو ایک خرابی ملے گی کیونکہ آپ نہیں جانتے کہ کون سا کالم استعمال کرنا ہے۔

4. تاریخ کے کالم کو اصل تاریخوں میں تبدیل کریں۔

5. اپنے ڈیٹا میں تازہ ترین تاریخ تلاش کریں۔

6. کل سے موازنہ کریں۔

7. اگر آپ کا ڈیٹا پہلے سے ہی اپ ٹو ڈیٹ ہے۔

8. اگر آپ کا ڈیٹا پیچھے ہے۔

def determine_update_range(existing: pd.DataFrame) -> Tuple[bool, Optional[str]]:
    """Determine whether an update is needed and from what date.

    Logic:
    1. Check if existing data covers yesterday's date
    2. If yes → no update needed
    3. If no → start fetching from day after last data

    Returns:
        Tuple[should_update, start_date]
    """
    # Case 1: nothing on disk yet
    if existing.empty:
        print("No existing data found. Will fetch all data from earliest date.")
        return True, STATION_CONFIG.earliest_date

    if "date_obs_elab" in existing.columns:
        record_colname = "date_obs_elab"
    elif "record_date" in existing.columns:
        record_colname = "record_date"
    else:
        raise KeyError("Missing date column: expected 'date_obs_elab' or 'record_date'")

    s = pd.to_datetime(existing[record_colname], errors="coerce")
    last_day = s.max().date()
    yesterday = date.today() - timedelta(days=1)

    # Case 2: already current
    if last_day >= yesterday:
        print("nDataset already covers yesterday or later. No update needed.")
        return False, None

    # Case 3: fetch the gap
    next_day = (last_day + pd.Timedelta(days=1))
    print(f"nWill retrieve data starting from: {next_day}")
    return True, next_day.isoformat()

یہاں چند باتیں قابل توجہ ہیں۔

سب سے پہلے، فنکشن چیک کرتا ہے: کالم کے دو ممکنہ نام: date_obs_elabRaw API نام یا پہلے سے نام تبدیل کر دیا گیا انگریزی نام record_date. صرف ایک چیز کو فرض نہ کریں۔ یہ اس بات کو یقینی بناتا ہے کہ فنکشن کام کرے گا اس سے قطع نظر کہ آپ اسے نئے درآمد شدہ خام ڈیٹا پر کال کرتے ہیں یا ڈسک سے بھری ہوئی پہلے سے پروسیس شدہ CSV پر۔

errors="coerce" یہ یہاں ظاہر ہوتا ہے اور دوبارہ ظاہر ہوگا (ہم اسے مزید تفصیل سے حصہ 5 میں دیکھیں گے)۔ پانڈا جن تاریخوں کو پارس نہیں کر سکتے ان میں شامل ہیں: NaT (وقت نہیں) ایک استثناء کو بڑھانے کے بجائے۔

واپسی کی اقسام یہ ہیں: Tuple[bool, Optional[str]]. یہ ٹیپل دو متعلقہ نتائج کو گروپ کرتا ہے۔ اپ ڈیٹ کرنے کی ضرورت ہے؟ مجھے کب اپ ڈیٹ کرنا چاہیے؟ یہ دو الگ الگ اقدار، یا ایک مبہم قدر واپس کرنے سے بہتر ہے جس کا مطلب سیاق و سباق کے لحاظ سے مختلف چیزیں ہیں۔

بہترین طرز عمل: استعمال کریں Tuple واپسی کی قسم (یا مزید فیلڈز کے لیے ایک چھوٹی ڈیٹا کلاس، یا NamedTuple) متعلقہ نتائج کو ایک ساتھ گروپ کریں۔ واضح طور پر دستاویز کریں کہ ہر پوزیشن کا کیا مطلب ہے۔ اگر آپ کچھ مختلف طریقے سے واپس کرتے ہیں قسم دستاویزات کے بغیر مختلف کوڈ پاتھ کا استعمال کنفیوژن اور کیڑے کا ایک عام ذریعہ ہے۔ "اس میں کیا حرج ہے؟ None "کبھی یہ ایک تار ہے اور کبھی یہ ایک تار ہے؟".

تین منظرناموں کے لیے ایک فوری اسکین چلائیں:

# Test 1: No existing data
should_update, start_date = determine_update_range(pd.DataFrame())
# → True, "1900-01-01"

# Test 2: Old existing data (covers only Jan 10-15)
# → True, "2026-01-16"  (the day after the last known date)

# Test 3: Recent data that already covers yesterday
# → False, None

API تخروپن

ڈیزائن منطق

پیداوار میں، بازیافت کا مطلب ایک حقیقی HTTP کال ہے۔

requests.get(
    "https://hubeau.eaufrance.fr/api/v2/hydrometrie/obs_elab",
    params={"code_entite": "F700000109", "size": 20000, ...}
)

اصل API کالیں حقیقی مسائل لاتی ہیں۔ اس ٹیوٹوریل میں، ہم اسے پہلی بار پڑھنے سے پہلے ہر چیز کی تعمیر اور جانچ کریں گے۔ جعلی جنریٹر یہ ڈیٹا اسی فارمیٹ میں واپس کرتا ہے جیسا کہ اصل API۔ منطق کے کام کرنے کے بعد ہی انہیں اصل اختتامی نقطہ پر تبدیل کیا جاتا ہے۔

یہ تکنیک اس منصوبے سے آگے مفید ہے۔ سب سے پہلے، اپنے فکسچر یا فرضی آئٹم کے لیے تبدیلی کی منطق بنائیں اور جانچیں۔ اس طرح آپ "کیا میری تجزیہ غلط ہے؟" میں نہیں پھنسیں گے۔ اور "کیا آپ کا نیٹ ورک اب غیر مستحکم ہے؟" ایک ہی وقت میں.

کوڈ لیول واک تھرو

generate_mock_api_data() مندرجہ ذیل سے شروع کرتے ہوئے، ہم اسٹیشن کے لیے روزانہ ایک ریکارڈ کے ساتھ جعلی نمونے کا ڈیٹا تیار کرتے ہیں۔ start_date.

یہ کیسے کام کرتا ہے:

• یہ بدلتا ہے start_date ایک حقیقی تاریخ پر۔

• اگلی بار کے لیے دہرائیں۔ num_days.

• ہم ہر روز ایک فرضی مشاہداتی لغت بناتے ہیں۔

• ڈیٹا کو حقیقت پسندانہ بنانے کے لیے پانی کی سطح میں چھوٹی بے ترتیب تبدیلیاں شامل کریں۔

• حیثیت، معیار، اور طریقہ کار کے لیبل تصادفی طور پر منتخب کیے گئے ہیں۔

• یہ لغات کی فہرست لوٹاتا ہے۔

def generate_mock_api_data(station_code: str, start_date: str, num_days: int = 10) -> List[Dict]:
    """Generate realistic mock API data for demonstration.

    Simulates what HubEau API would return: list of observation dicts.
    """
    start = pd.to_datetime(start_date).date()
    records = []

    validation_statuses = ["Donnée validée", "Donnée brute", "Donnée pré-validée"]
    qualities = ["Bonne", "Non qualifiée", "Douteuse"]
    methods = ["Mesurée", "Calculée", "Expertisée"]

    for i in range(num_days):
        obs_date = start + timedelta(days=i)
        base_level = 5500 + int(station_code[-2:])  # Varies by station
        noise = random.randint(-200, 200)
        water_level = base_level + noise

        record = {
            "code_site": "mock_" + station_code[1:],
            "code_station": "mock_" + station_code,
            "date_obs_elab": obs_date.isoformat(),
            "resultat_obs_elab": water_level,
            "date_prod": (obs_date + timedelta(days=1)).isoformat(),
            "code_statut": "1",
            "libelle_statut": random.choice(validation_statuses),
            "code_methode": "1",
            "libelle_methode": random.choice(methods),
            "code_qualification": "1",
            "libelle_qualification": random.choice(qualities),
            "longitude": 2.3522 + random.uniform(-0.01, 0.01),
            "latitude": 48.8566 + random.uniform(-0.01, 0.01),
            "grandeur_hydro_elab": "mock_HIXnJ",
        }
        records.append(record)

    return records

حقیقی ڈیٹا حاصل کریں۔

ڈیزائن منطق

دو افعال نکالنے کو سنبھالتے ہیں: انہیں جان بوجھ کر مندرجہ ذیل طور پر تقسیم کیا گیا تھا: واحد ذمہ داری کا اصول (SRP): ہر خصوصیت کو تبدیل کرنے کی ایک وجہ ہونی چاہیے۔

fetch_all_data() (orchestrator)
    ├── fetch_single_station_data(station_1) ← handles all complexity
    ├── fetch_single_station_data(station_2) ← handles all complexity
    └── fetch_single_station_data(station_n) ← handles all complexity

• fetch_single_station_data() مالک ہر گندا اسٹیشن کے لیے مخصوص پیچیدگیاں: صفحہ بندی، کرسر کی ترقی، اسٹاپ کنڈیشنز، اور نیٹ ورک ایرر ہینڈلنگ۔

• fetch_all_data() میں اس میں سے کسی کا مالک نہیں ہوں۔ یہ صرف اسٹیشنوں اور نمائندوں کے لیے دہرایا جاتا ہے۔

اس تقسیم کے دو انعامات ہیں۔ سب سے پہلے، آپ آرکیسٹریشن کوڈ کو چھوئے بغیر ایک اسٹیشن کی صفحہ بندی کی حکمت عملی کو ڈیبگ یا تبدیل کر سکتے ہیں۔ دوسرا، اگر آپ درآمد کو متوازی کرنا چاہتے ہیں ( concurrent.futures یا asyncioمثال کے طور پر) آرکیسٹریٹر صرف چھونے کی جگہ۔

کوڈ لیول واک تھرو

fetch_single_station_data() موک ٹیسٹ ڈیٹا یا اصلی API سے صفحہ کے لحاظ سے اسٹیشن ڈیٹا صفحہ حاصل کریں جب تک کہ آپ کے پاس اپنی ضرورت کی ہر چیز نہ ہو۔

• اگر use_mock=Trueیہ استعمال کیا جاتا ہے جعلی ڈیٹا اصل API کو کال کرنے کے بجائے۔

  1. یہ کال کرتا ہے generate_mock_api_data()

  2. نتیجہ کو ڈیٹا فریم میں تبدیل کریں۔

  3. تاریخ کے کالم کو پانڈا کی اصل تاریخوں میں تبدیل کریں۔

  4. متعلقہ ڈیٹا فریم لوٹاتا ہے۔

• اگر use_mock=Falseکہ اصل API کی درخواست:

  1. دوبارہ قابل استعمال HTTP سیشن بناتا ہے۔

  2. سے شروع کریں۔ start_date

  3. API سے بار بار ڈیٹا کے صفحات کی درخواست کرتا ہے۔

  4. یہ مندرجہ ذیل صورتوں میں رک جاتا ہے:

  5. تمام صفحات کو ایک ڈیٹا فریم میں یکجا کریں۔

  6. اگر کچھ حاصل نہیں کیا گیا تھا، تو یہ ایک خالی ڈیٹا فریم لوٹاتا ہے۔

def fetch_single_station_data(station_code: str, start_date: str, use_mock: bool = True) -> pd.DataFrame:
    """Fetch all hydrometric data for a single station from mock API (default) or from the real endpoint.

    In production (cursor based pagination strategy):
    - Fetches max_per_page records per request
    - Continues until no new data or yesterday's date reached
    - Stop when: no data returned | last date >= yesterday | page was not full
    - Handles network errors gracefully
    """
    if use_mock:
        data = generate_mock_api_data(station_code, start_date, num_days=7)
        page_df = pd.DataFrame(data)
        page_df["date_obs_elab"] = pd.to_datetime(
            page_df["date_obs_elab"], errors="coerce").dt.normalize()
        return page_df

    # Real implementation
    else:
        session = requests.Session()  # Reuse TCP connection across pages
        frames = []
        cursor = start_date

        while True:
            params = {
                "code_entite": station_code,
                "grandeur_hydro_elab": API_CONFIG.metric,
                "date_debut_obs_elab": cursor,
                "size": API_CONFIG.max_per_page,
            }

            try:
                response = session.get(
                    API_CONFIG.base_url,
                    params=params,
                    timeout=API_CONFIG.timeout_seconds  # Best practice, always set
                )
                response.raise_for_status()
            except requests.RequestException as e:  # Don't let one station kill the whole pipeline
                print(f"Error fetching data for station {station_code}: {e}")
                break

            data = response.json().get("data", [])
            if not data:  # Empty response: we've exhausted this station
                break

            page_df = pd.DataFrame(data)
            page_df["date_obs_elab"] = pd.to_datetime(
                page_df["date_obs_elab"], errors="coerce").dt.normalize()
            frames.append(page_df)

            last_page_date = page_df["date_obs_elab"].max()
            yesterday = date.today() - timedelta(days=1)

            # Prevent infinite loops
            if pd.isna(last_page_date) or last_page_date.date() >= yesterday:
                break

            cursor = (last_page_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")

            if len(data) < API_CONFIG.max_per_page:
                break

    if frames:
        return pd.concat(frames, ignore_index=True)
    return pd.DataFrame()

صفحہ بندی کے لوپ دراصل کیسے کام کرتے ہیں؟

آئیے قدم بہ قدم اس پر ایک نظر ڈالتے ہیں۔ یہ پورے لیپ ٹاپ میں سب سے گھنی منطق ہے۔

1. کو درخواست بھیجیں۔ date_debut_obs_elab=cursor: "براہ کرم مجھے اس تاریخ کا ریکارڈ دیں۔"

2. اگر درخواست مکمل طور پر ناکام ہوجاتی ہے (requests.RequestException)، اسے ریکارڈ کریں اور break. break لوپ کو فوری طور پر روکتا ہے اور جاری رہتا ہے۔ ایک اسٹیشن پر نیٹ ورک کا مسئلہ دوسرے تمام اسٹیشنوں پر پائپ لائن میں خلل نہیں ڈالنا چاہیے۔

3. اگر جواب میں کوئی ڈیٹا نہیں ہے تو، آپ نے پکڑ لیا ہے۔ break.

4. دوسری صورت میں، مندرجہ ذیل کو ذہن میں رکھیں: حالیہ اس صفحہ پر دکھائی گئی تاریخیں (last_page_date)۔

5. اگر تازہ ترین تاریخ پہلے سے موجود ہے۔ >= yesterdayآپ نے پکڑ لیا ہے: break.

6. بصورت دیگر، کرسر کو اس پر منتقل کریں۔ last_page_date + 1 day اگلے صفحے کے لیے دوبارہ دہرائیں۔

7. حفاظتی جال کے طور پر: اگر صفحہ واپس کر دیا جاتا ہے۔ کم ریکارڈ کے مقابلے میں max_per_pageاس کا مطلب یہ بھی ہے کہ انجام کو پہنچ گیا ہے۔ API جزوی صفحہ واپس نہیں کرے گا جب تک کہ اس کا ڈیٹا ختم نہ ہوجائے، اس لیے: break.

حتمی جانچ (مرحلہ 7) کلاسک طریقہ ہے۔ صفحہ بندی ختم کرنے کا جائزہ. یہ ہمیشہ ضروری نہیں ہوتا ہے۔ next_page API سے ٹوکن۔ اگر یہ پورا صفحہ ہے۔ size=20000 اور صرف تم واپس آؤ 4213 ریکارڈ پر نظر ڈالیں تو مزید کچھ نہیں لانا ہے۔

اسے تین مخصوص، جان بوجھ کر انتخاب کی ضرورت ہے:

session = requests.Session()

کوئی راستہ نہیں Session آبجیکٹ ایک ہی میزبان کو متعدد درخواستوں پر بنیادی TCP کنکشن کو دوبارہ استعمال کرتا ہے۔ یہ ہر ایک صفحے کی درخواست پر ایک نیا TCP/TLS ہینڈ شیک ہونے سے روکتا ہے۔ مختصر یہ کہ یہ آپ کے لیے تیز تر ہے اور حب ایو کے سرورز کے لیے زیادہ شائستہ ہے۔

بہترین طرز عمل: میں جب بھی فون کرتا ہوں۔ requests.get() ایک لوپ میں ایک ہی میزبان کے لیے ایک سے زیادہ بار Session.

except requests.RequestException as e:

RequestException تمام مستثنیات کے لیے بیس کلاس۔ requests آپ کو جن مسائل کا سامنا ہو سکتا ہے: ٹائم آؤٹ، کنکشن کی خرابیاں، HTTP کی خرابیاں raise_for_status() اور مزید۔ یہاں، بیس کلاس کو پکڑنے کا مطلب ہے کوئی بھی نیٹ ورک کے مسائل کو اسی نرمی سے نمٹا جاتا ہے۔ نوٹ کریں، اس اسٹیشن کو حاصل کرنا بند کریں، اور آگے بڑھیں۔

timeout=API_CONFIG.timeout_seconds

بنیادی طور پر requests فون کال کوئی ٹائم آؤٹ نہیں ہوتا ہے۔. واضح ٹائم آؤٹ کے بغیر، ہینگ سرور پوری پائپ لائن کو غیر معینہ مدت تک لٹکانے کا سبب بن سکتا ہے۔ درخواست کے اعلی درجے کے استعمال کے دستاویز میں کہا گیا ہے کہ بیرونی سرورز کی درخواستوں کے ساتھ ایک ٹائم آؤٹ منسلک ہونا چاہیے۔

بہترین طرز عمل: تمام بیرونی I/O کالوں کو لپیٹ دیتا ہے۔ try/except. ہمیشہ خوش اسلوبی سے ناکام رہیں۔ غلطی کو لاگ کریں اور پائپ لائن کو بحال کرنے یا جاری رکھنے کی اجازت دیں۔ یہ غیر مستحکم درخواستوں کو دن کے وقت کی کارروائیوں میں خلل ڈالنے سے روکتا ہے۔

اب آرکیسٹریٹر fetch_all_data()یہ جان بوجھ کر بہت آسان ہے۔

1. تمام ریورس کوڈز کو دہراتا ہے۔

2. ان کو ایک بڑے ڈیٹا فریم میں یکجا کریں۔

def fetch_all_data(start_date: str, use_mock: bool = True) -> pd.DataFrame:
    """Orchestrator: Fetch data for all configured stations."""
    frames = []

    for station_code in STATION_CONFIG.station_codes:
        print(f"Fetching data for station {station_code}...")
        df_station = fetch_single_station_data(station_code, start_date, use_mock=use_mock)

        if not df_station.empty:
            print(f"  Got {len(df_station)} records")
            frames.append(df_station)
        else:
            print(f"  (no data)")

    if frames:
        return pd.concat(frames, ignore_index=True)
    return pd.DataFrame()

بس۔ لوپ کے ساتھ pd.concat. وہ تمام مشکل سے جیتی گئی پیچیدگی ایس آر پی کے ذریعہ بیان کردہ بہت پرتوں کے نیچے موجود ہے۔

حصہ 5: تبدیلی کے مراحل

یہ وہ جگہ ہے جہاں نیا درآمد شدہ خام ڈیٹا قابل اشاعت ڈیٹا بن جاتا ہے۔ اس پورے حصے کا ڈیزائن فلسفہ حسب ذیل ہے: خالص تقریب. ہر ایک ڈیٹا فریم لیتا اور واپس کرتا ہے۔ نیا ڈیٹا فریم بغیر کسی ضمنی اثرات کے آؤٹ پٹ ہے۔ اسے ایک بڑا کام کرنے کے جال سے بچیں۔

    (EXTRACT)
    Raw API Data
    ↓
    (TRANSFORM)
1. Type parsing (datetime, numeric)
2. Column renaming (French → English)
3. Categorical mapping (validation status, quality)
4. Derived columns computation (flood alert flags)
5. Column reordering (logical grouping)
6. Sorting & index reset
    ↓
    (LOAD)
    Publication-ready dataset

اسے ایک بڑے فنکشن کے بجائے 6 چھوٹے مراحل میں کیوں توڑ دیں؟ ہر حصے کا تجربہ کیا جا سکتا ہے اور گھر میں تبدیل کیا جا سکتا ہے۔ اگر طے شدہ رن کے دوران صبح 3 بجے کچھ غلط ہو جاتا ہے، تو آپ ہر فنکشن کو انفرادی طور پر ڈیبگ کر سکتے ہیں۔ آپ اس بات کی نشاندہی کرسکتے ہیں کہ کون سے قدم نے غلط آؤٹ پٹ تیار کیا۔ ایک 200 لائن فنکشن کو منتخب کرنے کی ضرورت نہیں ہے۔

تجزیہ اور خوبصورت جبر ٹائپ کریں۔

ڈیزائن منطق

CSV یا JSON API میں ڈیٹا بطور ڈیفالٹ سٹرنگ کے طور پر شروع ہوتا ہے۔ پانڈا تاریخوں کو تاریخ کے مطابق ترتیب دیتے ہیں اور تاریخ کی ریاضی کرتے ہیں (last_date + timedelta(days=1)) یا عددی اقدار کا موازنہ کریں (water_level > 6000)۔ قسم کی مماثلت ایک عام غلطی ہے جو بیچ پائپ لائن کو توڑ دیتی ہے۔ اس طرح کی ایک غلط لائن: "N/A"ایک کٹی ہوئی تاریخ، ایک غلط ترتیب شدہ قدر، یا ایک غلط کریکٹر اور سخت تجزیہ کار ایک استثناء اٹھائے گا جو پورے عمل کو ختم کردے گا۔

بہترین طرز عمل: ذیل میں دونوں تبادلوں کے افعال استعمال کریں۔ errors="coerce"لہذا وہ اقدار جن کی پانڈا تجزیہ نہیں کرسکتے ہیں: NaT (وقت نہیں) یا NaN (نمبر نہیں) غلطی پھینکنے کے بجائے۔ اسے خوبصورت جبر کہا جاتا ہے۔

کوڈ لیول واک تھرو

convert_to_date() اور convert_to_numeric() یہ ایک چھوٹا مددگار فنکشن ہے جو چیک کرتا ہے کہ آیا کسی مخصوص کالم میں صحیح قسم ہے۔

کے لیے convert_to_date():

1. df.copy() چونکہ ہم ایک الگ کاپی بناتے ہیں، اصل جدول میں کوئی تبدیلی نہیں ہوتی۔

2. لوپ ہر کالم کے نام سے گزرتا ہے۔ columns

   • if col in df.columns تبادلوں کی کوشش کرنے سے پہلے چیک کریں کہ آیا کالم واقعی موجود ہے۔

   • pd.to_datetime(...) متن کو اس سے تبدیل کریں: "2026-07-12" پانڈا کی اصل تاریخ/وقت کی قدروں کے ساتھ

   • errors="coerce" اس کا مطلب ہے کہ یہ ایک غلط قدر ہوگی۔ NaT (گمشدہ تاریخ) غلطی کو بڑھانے کے بجائے:

   • .dt.normalize() وقت کا حصہ ہٹا دیں اور صرف آدھی رات کی تاریخ رکھیں۔

کے لیے convert_to_numeric():

1. df.copy() یہاں بھی استعمال ہوتا ہے۔

2. لوپ ہر کالم کے نام سے گزرتا ہے۔ columns

   • if col in df.columns تبادلوں کی کوشش کرنے سے پہلے چیک کریں کہ آیا کالم واقعی موجود ہے۔

   • لوپ کال pd.to_numeric() متن کو اس سے تبدیل کریں: "12.5" نمبروں سے

   • errors="coerce" بری قدر NaN گرنے کے بجائے

def convert_to_date(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """Convert specified columns to pandas datetime type."""
    df = df.copy()  # Never modify the original!
    for col in columns:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], errors="coerce").dt.normalize()
    return df


def convert_to_numeric(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """Convert specified columns to numeric (float) type."""
    df = df.copy()
    for col in columns:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df

یہ کیوں مفید ہے؟

• ڈیٹا کو جمع کیا جا سکتا ہے: گروپ، خلاصہ، وغیرہ...

• تاریخوں کو اب اصل تاریخ کے مطابق ترتیب اور فلٹر کیا جا سکتا ہے۔

• اعداد حسابات میں درست طریقے سے کام کرتے ہیں جیسے کہ اوسط، رقم اور موازنہ۔

• مخلوط اقسام کی وجہ سے پیدا ہونے والے کیڑوں سے بچیں جیسے: "12" اور 12.

دونوں pd.to_datetime اور pd.to_numeric سرکاری پانڈا خصوصیت ہے errors پیرامیٹرز پہلے سے طے شدہ طور پر، وہ پیرامیٹرز مندرجہ ذیل ترتیب دیئے گئے ہیں: "raise"غلط ان پٹ ہوتا ہے۔ دیگر اختیارات میں شامل ہیں: "coerce" (null سے بدلیں) یا "ignore" (اچھوا چھوڑ دیں)۔ پانڈا کا منظر to_datetime دستاویزات اور to_numeric پیرامیٹرز کی مکمل فہرست کے لیے دستاویزات دیکھیں۔

گندے ان پٹ کی مثالوں کے خلاف ٹیسٹ کریں۔

messy_df = pd.DataFrame({
    "date_obs_elab": ["2026-01-15", "2026-01-16", "not a date", None],
    "resultat_obs_elab": [5800.0, "5900", "N/A", None],
})

type_safe_df = convert_to_date(messy_df, ["date_obs_elab"])
type_safe_df = convert_to_numeric(type_safe_df, ["resultat_obs_elab"])

# "not a date"  → NaT
# "N/A"         → NaN
# Pipeline continues safely — nothing crashed.

بہترین طرز عمل: ایک بری لائن کو پوری پھانسی کو روکنے نہ دیں۔ ایک استثناء کو بڑھانے کے بجائے، یہ غلط ڈیٹا کو کالعدم کرنے پر مجبور کرتا ہے۔ اگر آپ کو بعد میں ڈیٹا کے معیار کی چھان بین کرنے کی ضرورت ہو تو الگ سے نولز کو جھنڈا لگائیں یا لاگ ان کریں۔

یہ جان بوجھ کر کیا گیا سمجھوتہ ہے۔ کے درمیان انتخاب تاثیر (پائپ لائن چلتی رہتی ہے) ختم سختی (یہ تمام غلط قطاروں کو فوراً پکڑ لیتا ہے)۔ یہ عام طور پر طے شدہ غیر حاضر کام کے لئے صحیح کال ہے۔

df.copy(): کنونشن کے مطابق تبدیلی

اوپر کے دو فنکشنز کے اوپری حصے کو دوبارہ دیکھیں۔ df = df.copy(). یہ ایک لائن شروع میں ظاہر ہوتی ہے: تمام تبادلوں کے افعال یہ پائپ لائن میں ہے، اور یہ کوئی اتفاق نہیں ہے۔

Python DataFrame حوالہ کے ذریعے پاس کردہ ایک تغیر پذیر شے ہے۔ جب کسی فنکشن میں ترمیم کی جاتی ہے۔ df اگر آپ پہلے اسے کاپی کیے بغیر اسے اپنی جگہ پر چھوڑ دیتے ہیں، تو یہ کال کرنے والے کے اصل ڈیٹا فریم کو بھی بدل دے گا۔ یہ ایک کلاسک ضمنی اثر ہے اور یہ واقعی الجھا دینے والے کیڑے پیدا کر سکتا ہے۔ کال .copy() پہلے کا مطلب یہ ہے کہ ہر فنکشن کا آؤٹ پٹ بالکل نیا آبجیکٹ ہے۔ کال کرنے والے کے ذریعے پاس کردہ ان پٹ کو برقرار رکھا جاتا ہے۔ اچھوت ہونے کی ضمانت ہے۔.

بہترین طرز عمل: ڈیٹا فریم کے ساتھ اس طرح سلوک کریں: ناقابل تبدیل ان پٹ. اس کی جگہ پر ترمیم کرنے کے بجائے ایک نیا ڈیٹا فریم لوٹاتا ہے۔ یہاں تک کہ اگر اس کی تھوڑی مقدار میں میموری یا CPU کی لاگت آتی ہے، ڈیبگنگ جیت تقریباً ہمیشہ ان پائپ لائنوں کے لیے قابل قدر ہوتی ہے جو انتہائی پیمانے پر کام نہیں کرتی ہیں۔

آخر میں، ایک خود کار طریقے سے پتہ چلنے والا سہولت فنکشن ان دو نچلے درجے کے فنکشنز کو لپیٹ دیتا ہے۔ یہ متن کے کالم کو اسکین کرتا ہے، اندازہ لگاتا ہے کہ آیا اس میں تاریخ یا نمبر ہے، اور اوپر مماثل پارس فنکشن کو کال کرتا ہے۔

auto_convert_columns() میں کوشش کر رہا ہوں ~ اندازہ لگائیں کہ کون سا کالم تاریخ یا نمبر ہے۔ یہ پھر خود بخود ان اقسام کو درست کرتا ہے۔

یہ کیسے کام کرتا ہے:

ہم دو خالی فہرستوں سے شروع کرتے ہیں۔

یہ ڈیٹا فریم کے ہر کالم سے گزرتا ہے۔ اگر کالم پہلے سے ہی ایک حقیقی تاریخ/وقت یا عددی قسم ہے تو اسے چھوڑ دیں۔ اگر کالم متن کی طرح ہے (object یا string)، 10 تک غیر خالی نمونے کی قدروں کو دیکھتا ہے۔

پہلے یہ اس قدر کو بطور تاریخ پڑھنے کی کوشش کرتا ہے۔

• اگر یہ کام کرتا ہے، تو کالم اس میں شامل کیے جائیں گے: datetime_cols

اگر نہیں، تو یہ اسے ایک نمبر کے طور پر پڑھنے کی کوشش کرتا ہے۔

• اگر یہ کام کرتا ہے، تو کالم اس میں شامل کیے جائیں گے: numeric_cols

آخر میں ہم تمام تاریخ کے کالموں کو اس میں تبدیل کرتے ہیں: convert_to_date(). پھر تمام عددی کالموں کو اس میں تبدیل کریں: convert_to_numeric()

def auto_convert_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Auto-detect and convert datetime and numeric columns to the correct type."""
    datetime_cols = []
    numeric_cols = []

    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            continue
        if pd.api.types.is_numeric_dtype(df[col]):
            continue

        if pd.api.types.is_object_dtype(df[col]) or pd.api.types.is_string_dtype(df[col]):
            sample = df[col].dropna().head(10)
            if len(sample) == 0:
                continue

            try:
                pd.to_datetime(sample, errors="raise", format="mixed")
                datetime_cols.append(col)
                continue
            except (ValueError, TypeError):
                pass

            try:
                pd.to_numeric(sample, errors="raise")
                numeric_cols.append(col)
                continue
            except (ValueError, TypeError):
                pass

    df = convert_to_date(df, datetime_cols)
    df = convert_to_numeric(df, numeric_cols)
    return df

اپنی طرف دھیان دو try/except یہاں استعمال شدہ بلاکس errors="raise". بس اس کے برعکس اوپر دی گئی زبردستی کی حکمت عملیوں میں سے کوئی بھی صرف چھوٹے معاملات میں کام کرے گی۔ .head(10) نمونہ ہر کالم کے.

یہ قسم ہے -سونگھنا یہ ایک غیر حتمی تبدیلی کا مرحلہ ہے۔ ٹیسٹ "کیا یہ کالم تاریخوں، نمبروں، یا دونوں کے طور پر ظاہر ہوتے ہیں؟" اس کے بعد سستے نمونے حقیقی اور محاوراتی تبادلوں میں پہنچائے جاتے ہیں۔ پوری گرمی convert_to_date() یا convert_to_numeric(). دو مختلف errors حکمت عملی، دو مختلف کام۔

دو طرفہ نقشہ سازی کا استعمال کرتے ہوئے اسکیما کی تبدیلی

ڈیزائن منطق

Hub'Eau API درج ذیل فرانسیسی کالم کے نام اور فرانسیسی کیٹیگریکل اقدار واپس کرتا ہے: code_station یا "Donnée validée". عالمی صارفین کے لیے ڈیٹا سیٹ انگریزی میں دستیاب ہونا چاہیے۔ کالم کے ناموں کو ان لائن تبدیل کرنے سے جہاں بھی یہ آسان ہو آپ کے پورے کوڈ بیس میں فرانسیسی اور انگریزی کے درمیان نقشہ سازی کو جاری رکھے گا۔ پھر ضرورت پڑنے پر اسے تبدیل کرنے کا کوئی طریقہ نہیں ہے۔

اصلاحات: وضاحت کریں ایک قابل اعتماد میپنگ یہ ماڈیول کے سب سے اوپر ہے اور اس سے باقی سب کچھ اخذ کرتا ہے۔

کوڈ لیول واک تھرو

API_TO_EN Map API فیلڈ کے نام ان کے انگریزی ترجمے کے لیے۔

# Primary mapping: French API columns to English column names
API_TO_EN = {
    "code_site": "location_code",
    "code_station": "station_code",
    "date_obs_elab": "record_date",
    "resultat_obs_elab": "water_level_mm",
    "date_prod": "data_production_date",
    "code_statut": "validation_status_code",
    "libelle_statut": "validation_status",
    "code_methode": "production_method_code",
    "libelle_methode": "production_method",
    "code_qualification": "quality_code",
    "libelle_qualification": "quality_assessment",
    "longitude": "longitude",
    "latitude": "latitude",
    "grandeur_hydro_elab": "hubeau_elab_code",
}

# Reverse mapping: English to French (computed automatically)
EN_TO_API = {v: k for k, v in API_TO_EN.items()}

دوسری لائن (EN_TO_API) صرف ایک شارٹ کٹ ہے۔ ہر کلید اور قدر API_TO_EN. پیشگی سمجھ اسے بناتی ہے۔ اہم ڈیزائن پوائنٹس ہیں: EN_TO_API اس کا انتظام ہاتھ سے نہیں ہوتا۔ ماخوذ. آئٹمز کو شامل کرتے، ہٹاتے یا نام تبدیل کرتے وقت API_TO_EN، EN_TO_API اگلی بار جب ماڈیول چلایا جائے گا تو یہ خود بخود اپ ڈیٹ ہو جائے گا۔

وہاں ہے بالکل ایک جگہ پورے کوڈ بیس میں جہاں اسکیما میں تبدیلیاں ہونی چاہئیں۔

واضح قدر (ساتھ ہی کالم کے ناموں کے ساتھ) ایک ہی سلوک حاصل کرتے ہیں۔

CATEGORICAL_MAPPINGS = {
    "validation_status": {
        "Donnée validée": "validated",
        "Donnée brute": "raw",
        "Donnée pré-validée": "pre-validated",
    },
    "quality_assessment": {
        "Bonne": "good",
        "Non qualifiée": "unqualified",
        "Douteuse": "dubious",
    },
    "production_method": {
        "Calculée": "calculated",
        "Mesurée": "measured",
        "Expertisée": "expert-reviewed",
    },
}

ذیل کا فنکشن کالم کے ناموں اور زمرے کی قدروں کو معمول پر لانے کے لیے ان نقشوں کو لاگو کرتا ہے۔

rename_to_english():

1. اگر ڈیٹا فریم خالی ہے، تو یہ فوری طور پر ایک کاپی واپس کر دیتا ہے۔

2. API ناموں سے انگریزی ناموں میں تبدیل کیے گئے کالموں کی فہرست بنائیں۔

3. یہ پھر نام تبدیل شدہ ڈیٹا فریم واپس کرتا ہے۔

rename_to_api_schema():

1. یہ خالی ڈیٹا کی ایک کاپی بھی واپس کرتا ہے۔

2. بالکل اس کے برعکس۔ انگریزی نام API کے ناموں میں لوٹ جاتے ہیں۔

3. ڈیٹا فریم لوٹاتا ہے۔

(اگر آپ کو API کے اصل فارمیٹ میں ڈیٹا واپس بھیجنے کی ضرورت ہو تو یہ مفید ہے)۔

apply_categorical_mappings():

1. اصل ڈیٹا فریم کی ایک کاپی بناتا ہے تاکہ اس میں کوئی تبدیلی نہ ہو۔

2. ہر کالم کے لیے CATEGORICAL_MAPPINGSمیپنگ کا استعمال کرتے ہوئے اقدار کو تبدیل کریں۔

3. ڈیٹا فریم لوٹاتا ہے۔

def rename_to_english(df: pd.DataFrame) -> pd.DataFrame:
    """Rename API column names to English schema names."""
    if df.empty:
        return df.copy()

    columns_to_rename = {}
    for src, dst in API_TO_EN.items():
        if src in df.columns and dst not in df.columns:
            columns_to_rename[src] = dst

    return df.rename(columns=columns_to_rename)

def rename_to_api_schema(df: pd.DataFrame) -> pd.DataFrame:
    """Rename English column names back to API schema names (defensive/reverse operation)."""
    if df.empty:
        return df.copy()

    columns_to_rename = {k: v for k, v in EN_TO_API.items() if k in df.columns}
    return df.rename(columns=columns_to_rename)

def apply_categorical_mappings(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    for col_name, mapping in CATEGORICAL_MAPPINGS.items():
        if col_name in df.columns:
            df[col_name] = df[col_name].map(mapping).fillna(df[col_name])
    return df

یہاں دو دفاعی عادتیں قابل توجہ ہیں۔

جزوی ان پٹ کے خلاف مضبوطی: نام تبدیل کرنے کے دونوں فنکشنز صرف ان کالموں کا نام تبدیل کرتے ہیں جو درحقیقت ان پٹ میں موجود ہوتے ہیں (if src in df.columns)۔

نام تبدیل کرنے کا فنکشن، جو فرض کرتا ہے کہ تمام میپ شدہ کالم ہمیشہ موجود ہوتے ہیں، اس وقت کریش ہو جائے گا جب اسے ڈیٹا فریم پر بلایا جائے گا جو جزوی ہے یا اس کی شکل مختلف ہے۔

.map(mapping).fillna(df[col_name]). Series.map میپنگ ڈکشنری میں پائی جانے والی تمام اقدار کو بدل دیتا ہے اور تمام اقدار کو بدل دیتا ہے۔ ~ نہیں dict میں ملا NaN. زنجیر .fillna(df[col_name]) بحالی کے فوراً بعد اصل ان تمام مقامات پر اقدار درج کریں جہاں کوئی میپنگ لاگو نہیں کی گئی ہے۔ اس طرح، غیر متوقع درجہ بندی کی قدریں خود بخود کالعدم ہونے کے بجائے بغیر تبدیلی کے گزر جاتی ہیں۔ یہ ایک لطیف لیکن اہم مضبوطی کا انتخاب ہے۔

اس کو بالکل دیکھو .map()پھر-.fillna() اسٹیک اوور فلو پر زیر بحث محاورات: پانڈاس نقشے اور کارکردگی کی ایپلی کیشنز۔

ایک فوری راؤنڈ ٹرپ ٹیسٹ اس بات کی تصدیق کرتا ہے کہ دو طرفہ نقشہ سازی دراصل کام کرتی ہے۔

sample_api_df = pd.DataFrame({"code_station": ["F700000109"], ...})
renamed_df = rename_to_english(sample_api_df)          # French → English
reversed_df = rename_to_api_schema(renamed_df)          # English → French
# reversed_df.columns.tolist() == sample_api_df.columns.tolist()  → True

سیلاب کی وارننگ کا حساب کتاب

add_derived_columns() یہ ایک ون لائنر ہے۔ صرف ریاضی کر رہے ہیں۔ flood_alert. اگر پانی کی سطح طے شدہ سیلاب کی حد سے اوپر ہے۔ Configنتیجہ یہ ہے۔ True. ورنہ نتیجہ یہ ہے: False.

def add_derived_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Add a computed columns based on raw data.
    """
    df = df.copy()

    if "water_level_mm" in df.columns:
        df["flood_alert"] = df["water_level_mm"] > STATION_CONFIG.flood_threshold_mm

    return df

کالم ترتیب

یہاں کچھ چھوٹی، لیکن صارفین کے لیے نظر آنے والی تفصیلات ہیں: اپنے ڈیٹا کے ساتھ ایک بار ڈیفالٹ کالم آرڈر کی وضاحت کریں اور اسے کہیں بھی دوبارہ استعمال کریں۔ COLUMN_ORDER ترجیحی کالم ترتیب کو محفوظ رکھتا ہے۔

order_columns() ڈیٹا فریم کو دوبارہ ترتیب دیتا ہے تاکہ وہ کالم پہلے آئے اور اضافی کالم ختم ہوں۔

COLUMN_ORDER = [
    # Primary identifiers & measurements
    "station_code", "record_date", "water_level_mm", "flood_alert",
    # Metadata about the observation
    "hubeau_elab_code", "data_production_date",
    "validation_status_code", "validation_status",
    "production_method_code", "production_method",
    "quality_code", "quality_assessment",
    # Geographic info (less important)
    "location_code", "longitude", "latitude",
]

def order_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Reorder columns to preferred order."""
    present_cols = [c for c in COLUMN_ORDER if c in df.columns]
    other_cols = [c for c in df.columns if c not in present_cols]
    return df[present_cols + other_cols]

other_cols حفاظتی جال کے طور پر کام کرتا ہے: تمام کالم واضح طور پر درج نہیں ہیں۔ COLUMN_ORDER یہ اب بھی آخری میں شامل ہے۔ وہ خاموشی سے حذف نہیں ہوتے۔

قواعد (ترجیح کے لحاظ سے):

• شناخت کنندہ اور کلیدی فیلڈز کو پہلے رکھیں۔

• سب سے اہم، اکثر استعمال ہونے والے، اور سب سے زیادہ قابل اعتماد کالم پہلے رکھیں۔

• متعلقہ فیلڈز کی گروپ بندی کر کے اپنے ٹیبلز کو قدرتی طور پر پڑھنے کو بنائیں۔

• اختیاری یا شاذ و نادر استعمال شدہ فیلڈز کو آخر تک دھکیلیں۔

نقل کرنا

ڈیزائن منطق

جاتے جاتے ڈپلیکیٹس کو ہٹا دیں۔ ایک بڑھتی ہوئی پائپ لائن میں، تاریخ کی حدود اور ریکارڈ اوورلیپ ہو سکتے ہیں۔

• چونکہ API ڈیٹا دیر سے آتا ہے یا بعد میں مکمل ہوتا ہے، ہو سکتا ہے آپ اسے اسی دن واپس حاصل کرنا چاہیں۔

• ایک ہی مشاہدہ صفحہ بند جواب کے دو مختلف صفحات پر ظاہر ہو سکتا ہے۔

ڈپلیکیشن کے بغیر، یہ منظرنامے وقت کے ساتھ ساتھ آپ کے ڈیٹاسیٹ میں ڈپلیکیٹ قطاریں بنانے کا سبب بن سکتے ہیں۔ یہ بھی ایک پائپ لائن بناتا ہے، جیسا کہ حصہ 1 میں ذکر کیا گیا ہے۔ بے حسی: ایک بار چلائیں یا پانچ بار چلائیں اور نتیجے میں ڈیٹا سیٹ وہی رہے گا۔

ترمیم کرنے کے لیے، آپ کو اس بات کی وضاحت کرنی چاہیے کہ ریکارڈ کیا بناتا ہے۔ منفرد. ایک جامع کلید کی وضاحت کریں۔

key = (station_code, observation_date, water_level_value)

یہ خاص امتزاج کیوں؟ جسمانی طور پر، ایک سینسر (station_code) ایک دن کا ہے (observation_dateروزانہ پڑھنے کی زیادہ سے زیادہ مقدار (water_level_mm) اور یہ پڑھنا منفرد ہونا چاہیے۔

• مختلف اسٹیشنوں کے ریکارڈز ظاہر ہے کہ ایک دوسرے کے ساتھ اوورلیپ نہیں ہوتے ہیں۔

• مختلف دنوں میں دو پڑھنے کو اوورلیپ نہیں کیا جاتا۔

• مزید لطیف نکتہ: ایک ہی نشریاتی ادارے کی رپورٹ ایک ہی صرف ایک دن کے لیے مختلف یہ وہ اقدار ہیں جنہیں الگ مشاہدہ سمجھا جاتا ہے، مثال کے طور پر درست یا نظر ثانی شدہ پیمائش، اور یہ ڈپلیکیٹ اقدار نہیں ہیں جنہیں خود بخود رد کر دیا جانا چاہیے۔

کوڈ لیول واک تھرو

create_dedup_key() ہر لائن کے لیے ایک منفرد ٹیکسٹ ID بنانے کے لیے تین حصوں کو انڈر سکور کے ساتھ جوڑیں۔

• اسٹیشن کوڈ

• تاریخ (جیسے YYYY-MM-DD)

• پانی کی سطح کی قدر

مثال کلید:

"F700000109_2024-01-15_5800.0"

def create_dedup_key(df: pd.DataFrame) -> pd.Series:
    """Create unique deduplication key from station, day, and water level value.

    Key format: "station_code_YYYY-MM-DD_value"
    Example: "F700000109_2024-01-15_5800.0"
    """
    parts = []

    if "code_station" in df.columns:
        parts.append(df["code_station"].astype(str))

    if "date_obs_elab" in df.columns:
        parts.append(df["date_obs_elab"].dt.strftime("%Y-%m-%d"))

    if "resultat_obs_elab" in df.columns:
        parts.append(df["resultat_obs_elab"].astype(str))

    if not parts:
        return pd.Series(index=df.index, dtype="object")

    return pd.Series(
        ["_".join(row) for row in zip(*parts)],
        index=df.index
    )

کلیدی عمارت کیسے کام کرتی ہے: parts یہ سیریز کی فہرست کے ساتھ ختم ہوتا ہے، ہر ایک اہم جزو (اسٹیشن، تاریخ، قدر) کے لیے ایک، ہر ایک ڈیٹا فریم کی لمبائی کے برابر۔

آخری لائن کو اندر سے پڑھیں۔

    return pd.Series(["_".join(row) for row in zip(*parts)], index=df.index)

zip(*parts) کالموں کی اس فہرست کو قطار وار ٹوپلز میں بدل دیتا ہے۔ یہ حاصل کرتا ہے (station_1, date_1, value_1)پھر (station_2, date_2, value_2)وغیرہ۔ فہرست کی تفہیم پھر ہر انڈر سکور شدہ قطار کو فی قطار ایک سٹرنگ کلید میں جوڑ دیتی ہے۔

یہ "کالم کی فہرست → زپ → قطار ٹوپل" محاورے کچھ بھی لکھے بغیر ایک سے اخذ کردہ سیریز میں متعدد سیریز کو یکجا کرنے کا ایک عام اور موثر طریقہ ہے۔ .apply(lambda row: ..., axis=1). قطار در قطار .apply ویکٹرائزڈ سٹرنگ آپریشنز کے مقابلے میں، یہ پانڈوں میں بہت سست ہے۔

اصل ڈپلیکیشن اس میں کی جاتی ہے: remove_duplicates(). اس سے قطاریں ہٹائیں: "نیا" (نیا درآمد شدہ ڈیٹا) پہلے سے موجود ہے۔ "موجودہ" (تاریخی اعداد و شمار)

مرحلہ وار:

1. اگر ایک میز خالی ہے، تو یہ صرف واپس آتا ہے. new.

2. قسم کو دونوں ڈیٹا فریمز پر لاگو کرتا ہے۔ auto_convert_columns() سب سے پہلے، تاریخوں اور نمبروں کا صحیح موازنہ کیا گیا ہے۔

3. دونوں جدولوں میں ہر قطار کے لیے ڈپلیکیشن کیز بنائیں۔ create_dedup_key().

4. اس بات کا تعین کریں کہ درآمد شدہ ڈیٹا میں کون سی کلیدیں موجودہ تاریخی ڈیٹا سیٹ میں ظاہر نہیں ہوتی ہیں اور صرف بالکل نئی قطاریں رکھیں۔

5. اصل کالموں کو محفوظ رکھتے ہوئے فلٹر شدہ نتائج لوٹاتا ہے۔ new.

def remove_duplicates(existing: pd.DataFrame, new: pd.DataFrame) -> pd.DataFrame:
    """Remove rows from 'new' that already exist in 'existing'."""
    # Short-circuit: if either is empty, no work to do
    if existing.empty or new.empty:
        return new.copy()

    # Parse types on both sides for a fair comparison
    existing_std = auto_convert_columns(existing)
    new_std = auto_convert_columns(new)

    # Build the keys
    existing_keys = set(create_dedup_key(existing_std).dropna())
    new_keys = create_dedup_key(new_std)

    # Boolean mask: True where the new row is genuinely new
    mask = ~new_keys.isin(existing_keys)

    # Index back into the ORIGINAL (non-standardized) 'new' to preserve all columns
    result = new.iloc[new_keys[mask].index].copy()
    return result

کارکردگی اور مضبوطی کی دو تفصیلات قابل توجہ ہیں۔

سب سے پہلے existing_keys ہے setنہیں list. ممبرشپ ٹیسٹ (in / .isin()) ازگر کے لیے set ہے اوہ (1) اس کی وجہ یہ ہے کہ وہ ہیش تلاش کا استعمال کرتے ہوئے جواب دیتے ہیں، جو اوسطاً تیز ہوتے ہیں۔ کیا یہ آئٹم پہلے ہی یہاں ہے؟ براہ راست ٹیسٹ list ہے کو): آپ کو آئٹم کے حساب سے آئٹم اسکین کرنا پڑتا ہے اور یہ آپ کے موجودہ ڈیٹا سیٹ کے بڑھنے کے ساتھ آہستہ ہوتا جاتا ہے۔

دسیوں ہزار قطاروں والے ڈیٹا سیٹس کے لیے، ہر پائپ لائن میں نظر آنے والے فرق نمایاں ہیں اور یہ کام کے لیے صحیح ڈیٹا ڈھانچہ کو منتخب کرنے کی ایک درسی کتاب کی مثال ہیں۔ فہرست پر ایک عمومی بحث اور رہنمائی دیکھیں اور Python میں تلاش کی کارکردگی کو ترتیب دیں۔ .isin() فلٹرنگ کی کارکردگی۔

سادہ اصول: استعمال کریں set جب فوری رکنیت کی تصدیق ضروری ہے۔ استعمال کریں list جب آپ آرڈر یا ڈپلیکیشن میں دلچسپی رکھتے ہیں۔

دوسرا new.iloc[new_keys[mask].index] دوبارہ اشاریہ۔ اصلتصدیق شدہ نہیں۔ new ڈیٹا فریمنہیں new_std.

کیوں auto_convert_columns() میں صرف اسے حاصل کرنے کے لیے بھاگا تھا۔ موازنہ کے لیے مستقل قسم. کال کرنے والا اب بھی کال کرنا چاہتا ہے۔ اصل یہ ہمیں ہر اس چیز کے لیے خام اقدار اور اسکیما واپس دیتا ہے جو ڈپلیکیشن سے بچ جاتی ہے۔

ثانوی حساب نہ لگائیں۔ اتفاق سے سچائی کا ذریعہ بنیں۔ یہ فیصلہ کرنے کے لیے کہ کیا رکھنا ہے یا ہٹانا ہے، ہمیشہ اپنی کاپی میں ترمیم کریں۔ ایک بار جب آپ اپنا فیصلہ کرلیں تو اسے اپنے اصل ڈیٹا پر لاگو کریں۔ یہ آپ کو بعد کے مراحل کے لیے اصل ماخذ کی اقدار کو محفوظ رکھنے کی اجازت دیتا ہے۔

سیدھے الفاظ میں:

1. پہلی ترمیم صرف موازنہ کے مقاصد کے لیے ہے۔

2. اصل ڈیٹا کو فلٹر کرنے کے لیے اس موازنہ کا استعمال کریں۔

3. بعد کے مراحل میں پروسیسنگ کے لیے اصل قطار کو بغیر تبدیلی کے لوٹاتا ہے۔

بہترین طرز عمل: ڈپلیکیشن کیز کو سادہ، مستحکم (غیر متغیر) اور ہمیشہ شارٹ سرکیٹ رکھیں ( if existing.empty or new.empty: return new.copy()) بھاری کاموں کو انجام دینے سے پہلے۔ یہ سستی حفاظتی شق مہنگی ڈپلیکیشن منطق کو چھوڑ دیتی ہے جب پراسیس کرنے کے لیے کوئی ڈیٹا نہ ہو۔

یہ سب ایک ساتھ رکھو postprocess()

مندرجہ بالا تمام حصے چھوٹے اور خود آزمائشی ہیں۔ آخری مرحلہ ہے۔ تبدیلی انہیں ایک ساتھ چپکائیں۔ ترتیب میںذیل میں ایک پائپ لائن فنکشن کے ساتھ:

def postprocess(df: pd.DataFrame) -> pd.DataFrame:
    """Apply all post-processing transformations."""
    if df.empty:
        return df

    df = df.copy()

    print("  1. Converting types...")
    df = auto_convert_columns(df)

    print("  2. Renaming columns (French → English)...")
    df = rename_to_english(df)

    print("  3. Mapping categorical values...")
    df = apply_categorical_mappings(df)

    print("  4. Adding derived columns...")
    df = add_derived_columns(df)

    print("  5. Reordering columns...")
    df = order_columns(df)

    print("  6. Sorting and resetting index...")
    df = df.sort_values(["record_date", "station_code"]).reset_index(drop=True)

    return df

اس فنکشن کی شکل پر غور کریں۔ یہ بنیادی طور پر ایک لکیری اسکرپٹ ہے جس میں چھ نمبر والے پرنٹ شدہ مراحل ہیں۔ ہر ایک پہلے سے طے شدہ خالص فنکشن کو کال کرتا ہے۔ کوئی نئی بات نہیں ہے۔ منطق بس یہاں ترتیب. یہ جان بوجھ کر ہے۔

بہترین طرز عمل: ہر قدم پر اپنی پیش رفت کو واضح طور پر ریکارڈ کریں۔ جب صبح 3 بجے طے شدہ کام ناکام ہوجاتا ہے، تو ایک واضح مرحلہ وار لاگ دو منٹ کی تشخیص اور ایک گھنٹے کے اندازے کے درمیان فرق پیدا کرتا ہے۔ ایک اچھا آپشن لاگر استعمال کرنا ہے۔ یہاں یہ ٹیوٹوریل کلاسک کنسول پرنٹنگ کا استعمال کرتا ہے (print(f" 1. Converting types..."))۔

حصہ 6: لوڈ فیز

ڈیزائن منطق

لوڈ آخری مرحلہ: پروسیس شدہ ڈیٹا لیں اور اسے اس کی منزل تک لے جائیں۔ مشترکہ اہداف میں ڈیٹا گودام، ڈیٹا لیکس، اور ڈیٹا بیس شامل ہیں۔

شائع کرنے سے پہلے، پائپ لائن دو چیزیں تیار کرتی ہے: آؤٹ پٹ فولڈر خود اور ایک میٹا ڈیٹا فائل جو ڈیٹاسیٹ کی وضاحت کرتی ہے۔

کوڈ لیول واک تھرو

create_output_dir() چیک کریں کہ آیا آؤٹ پٹ فولڈر موجود ہے اور اگر نہیں تو اسے بنائیں۔

• اگر use_mock ہے Trueیہ استعمال کیا جاتا ہے KAGGLE_CONFIG.mock_output_dirجعلی ڈیٹا کی ڈائرکٹری۔

• بصورت دیگر، یہ اگلی سیٹ کی اصل آؤٹ پٹ ڈائرکٹری کا استعمال کرتا ہے۔ KAGGLE_CONFIG.output_dir.

def create_output_dir(use_mock: bool = False) -> None:
    """Create output directory (and parents) if it does not already exist."""
    output_dir = KAGGLE_CONFIG.mock_output_dir if use_mock else KAGGLE_CONFIG.output_dir
    output_dir.mkdir(parents=True, exist_ok=True)
    # exist_ok=True: idempotent, safe to call multiple times

exist_ok=True یہ ایک چھوٹی لیکن اہم تفصیل ہے۔ اس کے بغیر، Path.mkdir() تاثر FileExistsError اگر ڈائریکٹری پہلے سے موجود ہے، تو یہ ہر بار موجود رہے گی جب اسے پہلی کے بعد چلایا جائے گا۔

ماحول exist_ok=True ڈائریکٹری بنائیں بے حسی: 100 بار کال کرنے سے وہی اثر پڑے گا جو 1 بار کال کرنے سے ہوگا۔ فائل سسٹمز میں، حصہ 5 میں نقل کی منطق کے پیچھے یہ "دوبارہ کرنے کے لیے محفوظ" خیال ہے۔

Kaggle API ڈیٹا پیکیج کی تفصیلات کی پیروی کرتا ہے۔ وضاحت درکار ہے۔ dataset-metadata.json CSV کے ساتھ ایک فائل بنائیں۔ یہ پیرس سیلاب ڈیٹا سیٹ کی تلاش اور دریافت کے لیے اہم ہے۔

def create_metadata(df: pd.DataFrame, config: KaggleConfig) -> Dict:
    """Generate Kaggle dataset metadata from DataFrame and config."""
    if df.empty or "record_date" not in df.columns:
        first_date = "unknown"
        last_date = "unknown"
    else:
        first_date = df["record_date"].min().strftime("%Y-%m-%d")
        last_date = df["record_date"].max().strftime("%Y-%m-%d")

    return {
        "title": config.title,
        "id": config.dataset_slug,
        "licenses": [{"name": config.license_name}],
        "keywords": config.keywords,
        "temporalCoverage": {"startDate": first_date, "endDate": last_date},
        "geospatialCoverage": config.geospatial_coverage,
        "updateFrequency": config.update_frequency,
    }

میٹا ڈیٹا ہے KaggleConfig ڈیٹا کلاس۔ براہ کرم اس کا حوالہ دیں۔ temporalCoverage شمار کیا جاتا ہے ڈیٹا سے ہی (df["record_date"].min()/.max()) ہارڈ کوڈ نہیں ہے۔ جب بھی پائپ لائن چلتی ہے، میٹا ڈیٹا میں تاریخ کی حد خود بخود دستی بک کیپنگ کے بغیر حقیقت کی عکاسی کرتی ہے۔

آخر کار publish_to_kaggle() اپنا اپ ڈیٹ کردہ ڈیٹاسیٹ Kaggle پر اپ لوڈ کریں۔ Kaggle CLI کو ایک بیرونی کمانڈ کے طور پر طلب کریں۔

فنکشن:

1. موجودہ وقت حاصل کریں اور اسے اس طرح ٹیکسٹ لیبل میں تبدیل کریں: Weekly update: 2026-07-13 14:30:00

2. درج ذیل Kaggle کمانڈ لکھیں:

   • ڈیٹا سیٹ کا نیا ورژن شائع کریں۔

   • براہ کرم درج ذیل فائلیں استعمال کریں۔ KAGGLE_CONFIG.output_dir

   • ایک پیغام منسلک کریں۔

   • ڈائرکٹری کے مواد کو کمپریس کریں۔

3. کمانڈ پرنٹ کرتا ہے تاکہ آپ دیکھ سکیں کہ کیا کیا جائے گا۔

4. اس کا استعمال کرتے ہوئے کمانڈ چلائیں: subprocess.run(...)

اگر اپ لوڈ کام کرتا ہے:

• یہ پرنٹ کرتا ہے Successfully published to Kaggle.

اگر یہ ناکام ہوجاتا ہے:

• غلطی کو پرنٹ کریں اور پھر غلطی کو چھپانے سے روکنے کے لیے اسے دوبارہ بڑھائیں۔

def publish_to_kaggle() -> None:
    """Publish the updated dataset to Kaggle using the Kaggle CLI."""
    timestamp = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
    message = f"Weekly update: {timestamp}"

    cmd = [
        "kaggle", "datasets", "version",
        "-p", str(KAGGLE_CONFIG.output_dir),
        "-m", message,
        "--dir-mode", "zip",
    ]

    print("Publishing to Kaggle...")
    print("Command:", " ".join(cmd))

    try:
        subprocess.run(cmd, check=True)
        print("Successfully published to Kaggle.")
    except subprocess.CalledProcessError as e:
        print(f"Error publishing to Kaggle: {e}")
        raise

publish_to_kaggle Python کے ذریعے Kaggle کمانڈ لائن انٹرفیس (CLI) کی درخواست کریں۔ subprocess ماڈیول یہ ایک Python پائپ لائن میں بیرونی کمانڈ لائن ٹولز کو چلانے کا معیاری طریقہ ہے۔

یہاں دو انتخاب ایک عام عادت کے طور پر اپنانے کے قابل ہیں، نہ کہ صرف Kaggle کے لیے۔

• cmd کے ساتھ بنایا گیا تھا۔ listیہ کبھی بھی مربوط تار نہیں ہوتا ہے۔ دلائل کی فہرست پاس کریں۔ subprocess.run شیل کال کو مکمل طور پر چھوڑ دیں۔ یہ شیل انجیکشن کے خطرات سے بچتا ہے اور خالی جگہوں یا خصوصی حروف پر مشتمل دلائل کو مناسب طریقے سے ہینڈل کرتا ہے۔ آفیشل دیکھیں subprocess سیکیورٹی کے تحفظات۔

• check=True یعنی، اگر Kaggle CLI ناکام ہو جاتا ہے، تو یہ غیر صفر کی حیثیت کے ساتھ نکل جاتا ہے۔ subprocess.run تاثر CalledProcessError خاموشی سے لوٹنے کی بجائے۔ بغیر check=Trueایک ناکام اشاعت بالکل بقیہ کوڈ میں کامیاب اشاعت کی طرح دکھائی دیتی ہے۔

حصہ 7: پائپ لائن اسمبلی مکمل کریں۔

عالمی ریہرسل (مذاق موڈ)

پوری پائپ لائن کو آزادانہ طور پر بنایا اور جانچا جاتا ہے۔ اب بلاکس کو ایک لکیری فنکشن میں جوڑنے کا وقت آگیا ہے۔ سب سے پہلے، اسے مکمل طور پر فرضی ڈیٹا پر چلائیں۔ اس طرح کچھ بھی شائع نہیں ہوتا۔ مظاہرے کے مقاصد کے لیے، مصنوعی ڈیٹا سے بھرا ہوا CSV مقامی طور پر محفوظ کیا جاتا ہے۔ mock_output فولڈر

def run_etl_pipeline(use_mock: bool = True) -> pd.DataFrame:
    """Run the complete ETL pipeline: Extract → Transform → Load to CSV."""

    # STEP 1: LOAD EXISTING DATA
    if use_mock:
        loaded_df = pd.DataFrame(mock_api_response)
        loaded_df['date_obs_elab'] = pd.to_datetime(loaded_df['date_obs_elab'])
        loaded_df = rename_to_english(loaded_df)
    else:
        loaded_df = load_csv(KAGGLE_CONFIG.input_csv)

    # STEP 2: DETERMINE UPDATE RANGE
    should_update, start_date = determine_update_range(loaded_df)
    if not should_update:
        return loaded_df   # already current: nothing more to do

    # STEP 3: EXTRACT (FETCH DATA)
    fetched_data = fetch_all_data(start_date, use_mock=use_mock)

    # STEP 4: TRANSFORM I - DEDUPLICATE
    deduped_fetched_data = remove_duplicates(loaded_df, fetched_data)

    # STEP 5: TRANSFORM II - COMBINE (MERGE WITH EXISTING)
    new_df_english = rename_to_english(deduped_fetched_data)
    merged_historical_and_new = pd.concat([loaded_df, new_df_english], ignore_index=True)

    # STEP 6: TRANSFORM III - POST-PROCESS
    processed_records = postprocess(merged_historical_and_new)

    # STEP 7: EXPORT (LOAD TO CSV)
    create_output_dir(use_mock=use_mock)
    output_path = KAGGLE_CONFIG.mock_output_filename if use_mock else KAGGLE_CONFIG.output_csv_path
    processed_records.to_csv(output_path, index=False, sep=",")

    return processed_records

سب سے اوپر کے سات مراحل ان فنکشنز سے مطابقت رکھتے ہیں جو ہم نے پہلے ہی اس ٹیوٹوریل میں بنائے اور جانچے ہیں۔ ان کو جمع کرنا تقریباً مکینیکل ہے۔ یہ ہمیشہ چھوٹے، واحد ذمہ داری کے افعال کو منظم کرنے کا ایک فائدہ ہے۔

loaded_df                 = load_csv(...)                          # 1. load existing data
should_update, start_date = determine_update_range(...)            # 2. check what's needed
fetched_data               = fetch_all_data(start_date)             # 3. fetch new records
deduped_fetched_data       = remove_duplicates(existing, new_raw)   # 4. deduplicate
merged_historical_and_new  = pd.concat([existing, new_clean])       # 5. merge
processed_records          = postprocess(merged_historical_and_new) # 6. postprocess
processed_records.to_csv(...)                                       # 7. save
write_metadata() + publish_to_kaggle()                               # 8. publish

ہر سطر کا مقصد صرف بائیں سے دائیں پڑھنے سے واضح ہو جاتا ہے۔ یہ وہ مقام ہے جہاں گلنا آسان ہو جاتا ہے۔

main(): پائپ لائن آرکیسٹریشن

main() یہ موصل ہے۔ پہلے سے بنائے گئے تمام اجزاء کو صحیح ترتیب میں جوڑیں۔ postprocess() ایک درجے نیچے چلا گیا۔

def main() -> None:
    """Execute the full Paris Flood Monitoring ETL pipeline.

    EXTRACT
    1. Load the existing dataset from the Kaggle input mount
    2. Determine whether an update is needed (and from what date)
    3. Fetch new data from the Hub'Eau API

    TRANSFORM
    4. Deduplicate against the existing dataset
    5. Combine and post-process (type parsing, translation, derived cols...)

    LOAD
    6. Write the updated CSV and metadata file
    7. Publish the new dataset version to Kaggle

    Exits early if the dataset is already up to date.
    """
    final_dataset = run_etl_pipeline(use_mock=False)
    final_dataset.to_csv(KAGGLE_CONFIG.output_csv_path, index=False)
    write_metadata(final_dataset)

    print("n[FINAL STEP] Publishing to Kaggle...")
    publish_to_kaggle()

    print("Running post-run validation")
    validate_and_analyze(final_dataset)


if __name__ == "__main__":
    main()

کہ if __name__ == "__main__": باؤنسر

if __name__ == "__main__":
    main()

یہ Python کے سب سے عام محاوروں میں سے ایک ہے، اور یہ سمجھنا اچھا ہے کہ یہ کیا کرتا ہے۔ Python کی سرکاری دستاویزات کے مطابق: __main__:

بہترین طرز عمل: اپنے داخلی مقامات کو ہمیشہ اس طرح محفوظ رکھیں۔ ہم محفوظ طریقے سے ماڈیول تیار کرتے ہیں۔ درآمد کے لیے دستیاب ہے۔ صرف درآمد کر کے، آپ پوری پائپ لائن کو متحرک کیے بغیر انفرادی خصوصیات کی جانچ کر سکتے ہیں یا انہیں دوسری اسکرپٹس میں دوبارہ استعمال کر سکتے ہیں (بشمول Kaggle پر اصل اشاعت)۔

حصہ 8: لائیو سٹریمنگ اور حقیقی APIs کی طرف جانا

مذکورہ بالا سبھی فرضی ڈیٹا پر محفوظ طریقے سے چلتا ہے۔ اس کے بجائے، اپنی پائپ لائن کو اصل Hub'Eau API کی طرف اشارہ کرنے کے لیے:

1. براہ کرم اس کا حوالہ دیں۔ use_mock=False اوپر کی ترتیبات۔ یہ گھس جاتا ہے APIConfig.use_mockاور پھر ایک کال run_etl_pipeline(use_mock=False) اور main().

2. یقینی بنائیں KAGGLE_CONFIG.dataset_slug اور input_csv اشارہ کریں آپ کا اپنا Kaggle ڈیٹاسیٹ کو استعمال کرنے سے پہلے اسے کاپی کریں۔ آپ ڈیٹا سیٹس کے صرف نئے ورژن شائع کر سکتے ہیں جو آپ کے پاس ہیں۔ پھر، نوٹ بک اور ڈیٹاسیٹ دونوں کو فورک کریں۔ KaggleConfig پارٹ ٹائم نوکری.

3. اگر آپ کال کرنا چاہتے ہیں، تو براہ کرم Kaggle CLI کے ساتھ انسٹال اور تصدیق کریں۔ publish_to_kaggle() Google Colab یا Kaggle Notebooks کے باہر۔

باقی سب کی ضرورت ہے کوئی تبدیلی نہیں.

پھانسی کے بعد تصدیق کریں۔

validate_and_analyze() یہ لوپ کو بند کر دیتا ہے اور یہ صرف ایک اچھی چیز سے زیادہ ہے۔ یہ چلنا خوش آئند سمجھداری کی جانچ ہے۔ ~ بعد پائپ لائن مکمل ہو چکی ہے۔ آؤٹ پٹ ایک انسانی پڑھنے کے قابل رپورٹ ہے جس میں ظاہری شکل، ڈیٹا کی قسم، صفر شمار، اور خلاصہ کے اعداد و شمار شامل ہیں۔ water_level_mmسیلاب کی وارننگ ریکارڈز کی تعداد، عارضی کوریج اور فی اسٹیشن ریکارڈز کی تعداد۔

ڈیٹا تبدیل نہیں ہوتا ہے۔ یہ اس لیے موجود ہے کہ کوئی بھی، یا کوئی بھی مانیٹرنگ سسٹم، رن کے لاگ آؤٹ پٹ کو پڑھ کر فوراً دیکھ سکتا ہے کہ آیا اس ہفتے کے نمبر نارمل ہیں۔ براہ راست CSV کا تجزیہ کرنے کی ضرورت نہیں ہے۔ اسے آزمائیں!

خلاصہ

تم صرف واٹر ٹائٹ ایک ETL پائپ لائن جو کسی بھی چیز کو روکے بغیر بار بار چل سکتی ہے۔ کنکال یہاں ہے، اور تقریباً تمام طے شدہ ڈیٹا آپریشنز کے لیے ایک ہی پیٹرن کو دہرایا جاتا ہے۔ اس کا مطلب ہے APIs، فیلڈ میپنگ اور اہداف کو تبدیل کرنا۔

اس طرح کے مزید سبق کے لیے، میرے GitHub یا Kaggle پروفائلز کو دیکھیں۔

پڑھنے کے لیے شکریہ!

حوالہ جات