Build a complete, bounded reporting program

Trace the pipeline before running it

The complete program below is saved as reconcile_inventory.py. Its stages are read, validate, index, compare, summarize and export. Collection is outside the program. Remediation is outside the program. Each function owns a clear part of the contract, which makes it possible to test comparison without filesystem access.

The output directory is created exclusively after both inputs have validated. An existing directory causes failure. This avoids silently overwriting yesterday's evidence, and it prevents input validation errors from leaving an apparently useful empty report. Use a new output directory name for each run.

"""Offline, read-only inventory comparison. No endpoint or directory changes."""
import argparse
import csv
import hashlib
import io
import json
import re
import sys
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

FIELDS = ['asset_id', 'owner_id', 'site']
OUTPUT_FIELDS = ['asset_id', 'finding', 'field', 'registered', 'observed']


def parse_capture(value):
    if not isinstance(value, str):
        raise ValueError('collected_at must be a timezone-aware timestamp string')
    stamp = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if stamp.tzinfo is None or stamp.utcoffset() is None:
        raise ValueError('collected_at must include an offset')
    return stamp.astimezone(timezone.utc)


def normalize_row(row, record):
    if None in row or any(row.get(key) is None for key in FIELDS):
        raise ValueError(f'Record {record}: wrong field count')
    asset = row['asset_id'].strip().upper()
    owner = row['owner_id'].strip()
    site = row['site'].strip().upper()
    if not re.fullmatch(r'A-[0-9]{3}', asset):
        raise ValueError(f'Record {record}: invalid asset_id')
    if not re.fullmatch(r'u[0-9]{3}', owner):
        raise ValueError(f'Record {record}: invalid owner_id')
    if site not in {'HQ', 'BRANCH'}:
        raise ValueError(f'Record {record}: unsupported site')
    return {'asset_id': asset, 'owner_id': owner, 'site': site}


def unique_manifest(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('Manifest has a duplicate property')
        result[key] = value
    return result


def read_inventory(csv_path, manifest_path):
    source_bytes = csv_path.read_bytes()
    manifest_bytes = manifest_path.read_bytes()
    manifest = json.loads(manifest_bytes.decode('utf-8-sig'), object_pairs_hook=unique_manifest)
    if not isinstance(manifest, dict) or set(manifest) != {'complete', 'row_count', 'collected_at'}:
        raise ValueError('Manifest must contain complete, row_count and collected_at')
    if manifest['complete'] is not True:
        raise ValueError('Collector has not declared a complete export')
    if type(manifest['row_count']) is not int or manifest['row_count'] < 0:
        raise ValueError('Manifest row_count must be a non-negative integer')
    capture = parse_capture(manifest['collected_at'])
    reader = csv.DictReader(io.StringIO(source_bytes.decode('utf-8-sig'), newline=''), strict=True)
    if reader.fieldnames != FIELDS:
        raise ValueError('CSV header must be asset_id,owner_id,site in that order')
    indexed = {}
    for record, raw in enumerate(reader, start=2):
        row = normalize_row(raw, record)
        asset = row['asset_id']
        if asset in indexed:
            raise ValueError(f'Record {record}: duplicate normalized identity {asset}')
        indexed[asset] = row
    if len(indexed) != manifest['row_count']:
        raise ValueError('Parsed record count does not match the manifest')
    evidence = {
        'csv_path': str(csv_path.resolve()),
        'manifest_path': str(manifest_path.resolve()),
        'csv_sha256': hashlib.sha256(source_bytes).hexdigest(),
        'manifest_sha256': hashlib.sha256(manifest_bytes).hexdigest(),
        'collected_at': capture.isoformat(),
        'row_count': len(indexed),
    }
    return indexed, evidence


def compare(registered, observed):
    findings = []
    for asset in sorted(set(registered) | set(observed)):
        if asset not in observed:
            findings.append({'asset_id': asset, 'finding': 'REGISTER_ONLY', 'field': '', 'registered': 'present', 'observed': 'absent'})
        elif asset not in registered:
            findings.append({'asset_id': asset, 'finding': 'OBSERVED_ONLY', 'field': '', 'registered': 'absent', 'observed': 'present'})
        else:
            for field in ['owner_id', 'site']:
                if registered[asset][field] != observed[asset][field]:
                    findings.append({'asset_id': asset, 'finding': 'FIELD_MISMATCH', 'field': field, 'registered': registered[asset][field], 'observed': observed[asset][field]})
    return findings


def build_report(register_csv, register_manifest, observed_csv, observed_manifest, output):
    registered, register_evidence = read_inventory(register_csv, register_manifest)
    observed, observed_evidence = read_inventory(observed_csv, observed_manifest)
    findings = compare(registered, observed)
    common = set(registered) & set(observed)
    same = sum(registered[key] == observed[key] for key in common)
    counts = Counter(row['finding'] for row in findings)
    summary = {
        'schema': 1, 'status': 'complete',
        'generated_at': datetime.now(timezone.utc).isoformat(),
        'registered': register_evidence, 'observed': observed_evidence,
        'registered_assets': len(registered), 'observed_assets': len(observed),
        'common_assets': len(common), 'matching_common_assets': same,
        'mismatching_common_assets': len(common)-same,
        'field_mismatch_rows': counts['FIELD_MISMATCH'],
        'register_only_assets': counts['REGISTER_ONLY'],
        'observed_only_assets': counts['OBSERVED_ONLY'],
        'finding_rows': len(findings),
    }
    # Exclusive directory creation reserves the result location. If interrupted,
    # the directory may remain partial; consumers require a valid final summary.
    output.mkdir()
    with (output / 'findings.csv').open('x', encoding='utf-8', newline='') as stream:
        writer = csv.DictWriter(stream, fieldnames=OUTPUT_FIELDS)
        writer.writeheader()
        writer.writerows(findings)
    summary['findings_sha256'] = hashlib.sha256((output / 'findings.csv').read_bytes()).hexdigest()
    with (output / 'summary.json').open('x', encoding='utf-8') as stream:
        json.dump(summary, stream, indent=2)
        stream.write('\n')
    return summary


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('register_csv', type=Path)
    parser.add_argument('register_manifest', type=Path)
    parser.add_argument('observed_csv', type=Path)
    parser.add_argument('observed_manifest', type=Path)
    parser.add_argument('output_directory', type=Path)
    args = parser.parse_args()
    try:
        result = build_report(args.register_csv, args.register_manifest, args.observed_csv, args.observed_manifest, args.output_directory)
    except (OSError, ValueError, csv.Error) as error:
        print(f'Report failed: {error}', file=sys.stderr)
        return 2
    print(f"Complete comparison: {result['finding_rows']} finding rows; no source changes made")
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

Run it from the practice directory:

python reconcile_inventory.py register.csv register-manifest.json observed.csv observed-manifest.json result-baseline

A successful run prints four finding rows and exits with status 0. A handled validation or file error prints a diagnostic to standard error and exits with status 2. If your system uses python3 for the selected interpreter, use that command consistently. Keep the output directory name literal for the first run so you can inspect precisely where files were created.

Open result-baseline/findings.csv and result-baseline/summary.json. Compare them with your manual prediction, not just with the program's printed count. The findings must identify two A-002 field mismatches, A-004 as register-only and A-005 as observed-only. No finding should be issued for A-001 or A-003.

Understand the source evidence in the summary

The summary records absolute source paths, hashes of the CSV and manifest bytes, UTC collection timestamps and record counts. An absolute path helps reproduce the input selection within this environment. A hash identifies the exact bytes used. Neither proves who authored the file or that an upstream collector was authorized. A signed export or a controlled collection process would address different provenance questions.

Keep collection time separate from generation time. In the fixture the observed snapshot is five minutes later than the register snapshot. If A-002 changed owner during that interval, the mismatch might be a legitimate timing difference. The report identifies the mismatch; an analyst must evaluate the lifecycle evidence before deciding which value should change.

The summary's schema version helps a consumer reject an incompatible format rather than assume every future field has the same meaning. Names such as field_mismatch_rows and mismatching_common_assets deliberately distinguish output grain. A metric named errors would hide that distinction and encourage inaccurate comparisons.

Design for interrupted writes without claiming atomicity

The first version reserves the output directory and writes the findings before the final summary. If the process crashes or a disk error occurs, a partial directory may remain. Consumers must require a parseable final summary.json with the expected schema and completion status, then verify the expected files, counts and findings-file hash. The hash links the summary to the exported findings bytes; it is not a signed proof of the collector or its authority. A directory's existence is not proof of a successful run.

This is not a fully atomic publication protocol. Another process must not read files while this writer is active, and a robust production design may use a separate staging area, stronger integrity checks and an appropriate atomic publication mechanism. Do not claim that a quick rename is always safe across filesystems or that it always refuses replacement on every operating system.

If a failed attempt leaves a partial result directory, preserve it for diagnosis and use a new path for your next lab attempt. The script will not guess that an existing directory is disposable. This makes reruns inconvenient in a useful way: the learner must acknowledge the evidence instead of accidentally overwriting it.

Repeatability and idempotency are different

Run the same valid inputs with a different result directory. Findings and population counts should be equivalent and ordered consistently. The generation timestamp will differ, and paths can differ if you moved the inputs. Whole-summary byte equality is therefore not the expected test.

The program performs no remote mutation, so repeated comparisons cannot repeatedly change owners or retire assets. That is a bounded side-effect property. It is not proof of exactly-once execution in a scheduler. A later integration that creates tickets needs its own durable identity, duplicate handling and review contract.

Use the words precisely in your project README: the business comparison is deterministic for the same normalized input snapshots; report generation adds run metadata; outputs are not overwritten; source inventories are read-only.

Failure lab with expected outcomes

Change one input at a time and keep the baseline intact. First set the observed manifest to complete:false. Expect status 2, no result directory and no missing-device conclusion. Next duplicate the first asset with a different owner. Expect rejection before comparison; there must not be a “last row wins” report.

Change the observed timestamp to an offset-free string. Expect rejection. Change it instead to 2026-10-01T05:05:00-04:00: the reported UTC collection instant must remain 09:05 UTC. This test distinguishes formatting differences from different instants.

Run again with result-baseline as the output directory. Expect failure and unchanged report bytes. Finally point at a missing CSV path. Expect a diagnostic failure, not four register-only findings. A missing source file and an authoritative empty source are different cases.

For every attempt, record the predicted result, actual exit status, output existence and source hashes before and after. This evidence is more useful than a screenshot saying “script ran.”

Worked laboratory solution

Compare your reasoning with the worked solution

The repeated asset-1/Desk row can collapse under the explicit rule. asset-2 has conflicting types and cannot be resolved by silently keeping whichever row happens to come last. The batch must fail validation and preserve the old report. Write a rejection/conflict report separately, then require a corrected source or documented resolution. For a successful batch, create and validate a temporary output in an appropriate location before committing via the platform-supported replacement approach. Atomic rename behavior depends on filesystem boundaries and does not by itself guarantee every durability property.

Official reference for further study

Download the course workbook and practice materials