drdeveloper88's picture
Add usage code example (from_pretrained + BitsAndBytes + multilingual) to README
3b9fcc7
Raw
History Blame Contribute Delete
2.13 kB
from collections.abc import Iterable
from worlddisasterlm.data.schemas import DisasterRecord
class DisasterETL:
def collect_records(self) -> list[DisasterRecord]:
# Replace with API clients and ingestion jobs for production-scale collection.
return [
DisasterRecord(
source="ReliefWeb",
event_type="flood",
region="South Asia",
summary="Severe flooding displaced 12000 people and disrupted road access.",
severity="high",
),
DisasterRecord(
source="WHO",
event_type="epidemic",
region="East Africa",
summary="Localized cholera outbreak with urgent water sanitation requirements.",
severity="high",
),
DisasterRecord(
source="USGS",
event_type="earthquake",
region="Pacific Rim",
summary="Magnitude 6.8 earthquake with aftershock risk and infrastructure damage.",
severity="critical",
),
]
def deduplicate(self, records: Iterable[DisasterRecord]) -> list[DisasterRecord]:
seen: set[tuple[str, str, str, str]] = set()
deduped: list[DisasterRecord] = []
for record in records:
key = (record.source, record.event_type, record.region, record.summary)
if key not in seen:
deduped.append(record)
seen.add(key)
return deduped
def normalize(self, records: Iterable[DisasterRecord]) -> list[DisasterRecord]:
normalized: list[DisasterRecord] = []
for record in records:
normalized.append(
DisasterRecord(
source=record.source.strip(),
event_type=record.event_type.strip().lower(),
region=record.region.strip(),
summary=" ".join(record.summary.split()),
severity=record.severity.strip().lower(),
)
)
return normalized