| import argparse, csv, json |
| from pathlib import Path |
| try: |
| import yaml |
| except Exception: |
| yaml = None |
|
|
| def load_contract(path): |
| text = Path(path).read_text(encoding="utf-8") |
| if yaml: |
| return yaml.safe_load(text) |
| required = [] |
| for line in text.splitlines(): |
| if line.strip().startswith("- "): |
| required.append(line.strip()[2:]) |
| return {"required_fields": required, "quality_thresholds": {"completeness": 0.0}} |
|
|
| def validate(contract_path, file_path): |
| contract = load_contract(contract_path) |
| required = contract.get("required_fields", []) |
| with open(file_path, encoding="utf-8") as f: |
| rows = list(csv.DictReader(f)) |
| fields = rows[0].keys() if rows else [] |
| missing_fields = [f for f in required if f not in fields] |
| total_required = max(1, len(rows) * len(required)) |
| missing_values = sum(1 for row in rows for field in required if row.get(field, "") in ("", None)) |
| completeness = 1.0 - missing_values / total_required |
| schema_validity = 1.0 if not missing_fields else 0.0 |
| threshold = contract.get("quality_thresholds", {}).get("completeness", 0.0) |
| return { |
| "valid": not missing_fields and completeness >= threshold, |
| "row_count": len(rows), |
| "missing_fields": missing_fields, |
| "completeness": round(completeness, 4), |
| "schema_validity": schema_validity, |
| "quality_score": round(0.7 * completeness + 0.3 * schema_validity, 4), |
| } |
|
|
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--contract", required=True) |
| parser.add_argument("--file", required=True) |
| args = parser.parse_args() |
| print(json.dumps(validate(args.contract, args.file), indent=2)) |
|
|
