Implement a reviewed replacement and reconcile its report

Keep logic, batching and the command interface separate

classify_ticket validates one ticket and returns its label and reasons. evaluate manages population counts, row rejection and duplicate identity. The CLI parses the input, records its hash and publishes a new report. This separation lets tests target behavior without contacting anything or needing real system permissions.

The CLI rejects duplicate JSON properties and non-standard numeric constants rather than allowing the parser to silently choose a value. An input such as two ticket_id properties in one object is not the same as two clearly separate tickets. Ambiguous serialization should stop the input contract before analysis.

Save the reviewed program below as reviewed_classifier.py:

"""Classify fictional ticket fixtures; never contact or change a ticket system."""
import argparse
import hashlib
import json
import math
from pathlib import Path
import re
import sys

class ContractError(ValueError):
    pass

def number(value, field, positive=False):
    if type(value) not in (int,float) or (type(value) is float and not math.isfinite(value)):
        raise ContractError(field+' must be a finite numeric value, not text or bool')
    if not 0 <= value <= 168 or (positive and value == 0):
        raise ContractError(field+' outside approved lab range')
    return value

def classify_ticket(ticket):
    if not isinstance(ticket,dict) or set(ticket) != {'ticket_id','severity','age_hours','target_hours'}:
        raise ContractError('Unexpected ticket schema')
    if not isinstance(ticket['ticket_id'],str) or not re.fullmatch(r'T[0-9]{3}',ticket['ticket_id']):
        raise ContractError('Invalid ticket identity')
    if ticket['severity'] not in ('low','medium','high'):
        raise ContractError('Unsupported severity')
    age=number(ticket['age_hours'],'age_hours')
    target=number(ticket['target_hours'],'target_hours',positive=True)
    reasons=[]
    if ticket['severity']=='high': reasons.append('HIGH_SEVERITY')
    if age >= target: reasons.append('AT_OR_OVER_TARGET')
    return {'ticket_id':ticket['ticket_id'],'classification':'URGENT_CANDIDATE' if reasons else 'ROUTINE_CANDIDATE','reasons':reasons}

def unique_object(pairs):
    result={}
    for key,value in pairs:
        if key in result: raise ContractError('Duplicate JSON property')
        result[key]=value
    return result

def reject_constant(value):
    raise ContractError('Non-standard numeric JSON constant')

def evaluate(rows):
    if not isinstance(rows,list) or len(rows)>1000:
        raise ContractError('Expected an array of at most 1000 lab tickets')
    seen=set();results=[]
    for row in rows:
        identity=row.get('ticket_id') if isinstance(row,dict) else None
        valid_identity=isinstance(identity,str) and bool(re.fullmatch(r'T[0-9]{3}',identity))
        if valid_identity:
            if identity in seen: raise ContractError('Duplicate ticket identity')
            seen.add(identity)
        try: results.append(classify_ticket(row))
        except ContractError as error:
            results.append({'ticket_id':identity if valid_identity else 'unavailable','classification':'REJECTED','reasons':[str(error)]})
    return {'scope':'Candidate labels in supplied fictional data; no operational ticket action.',
            'input_records':len(rows),'urgent_candidates':sum(r['classification']=='URGENT_CANDIDATE' for r in results),
            'routine_candidates':sum(r['classification']=='ROUTINE_CANDIDATE' for r in results),
            'rejected_records':sum(r['classification']=='REJECTED' for r in results),'results':results}

def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('source',type=Path);parser.add_argument('output',type=Path);args=parser.parse_args()
    try:
        if args.source.resolve()==args.output.resolve(): raise ContractError('Source and output must differ')
        data=args.source.read_bytes()
        if len(data)>1000000: raise ContractError('Input exceeds lab byte budget')
        rows=json.loads(data,object_pairs_hook=unique_object,parse_constant=reject_constant)
        report=evaluate(rows);report['source_sha256']=hashlib.sha256(data).hexdigest()
        with args.output.open('x',encoding='utf-8') as stream:
            json.dump(report,stream,indent=2);stream.write('\n')
    except (OSError,ValueError) as error:
        print('Review failed: '+str(error),file=sys.stderr);return 2
    print(f"Classified {report['input_records']} records: {report['urgent_candidates']} urgent, {report['routine_candidates']} routine, {report['rejected_records']} rejected")
    return 0

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

Run the baseline with literal input

Use the supplied tickets.json. Its five rows are exactly the matrix above. Run:

python reviewed_classifier.py tickets.json review.json

Expect Classified 5 records: 2 urgent, 1 routine, 2 rejected. Inspect review.json: T002 has AT_OR_OVER_TARGET; T003 has HIGH_SEVERITY; the rejected rows explain their invalid data. The source hash identifies the exact input bytes used for this run. It does not establish that the input was authentic or complete.

Rerun with the same output path. Expect exit status 2 and the original report unchanged. Try using the input path as output and confirm refusal. Use a new name for each successful experiment. Retain the failed invocation and diagnostic message as evidence rather than deleting the old result to force success.

Understand the resource and publication limits

The CLI reads the local source before checking the one-million-byte lab budget. That limit is an admission check after reading, not a hard memory bound against a hostile oversized file. A production reader would enforce limits while streaming and account for parser expansion and nested data. The 1,000-record limit also is a lab contract, not a benchmark for unlimited automation.

Exclusive output opening prevents overwrite, but the JSON is written directly to that new file. A write failure or crash can leave a partial new report. A final filename alone is not proof of successful completion. An operational publisher would need a complete-write marker or an atomic publication design, integrity verification and a clear failed-run procedure.

The program has no business-hours calculation, tenant connection, authentication, durable ticket revision state or update idempotency. A separate live collector must establish source scope, coverage and timing semantics. Do not adapt the candidate report into automatic priority changes by adding a single API call at the end.

Review a proposed “helpful” improvement

Suppose an assistant proposes accepting strings such as "4" and interpreting an empty string as zero. Decide whether this is a parsing feature or a silent measurement policy. If you intentionally support numeric strings later, define syntax, decimal conventions, whitespace, units, invalid values and provenance. Test them independently and disclose the changed interface.

Another proposal changes URGENT_CANDIDATE to BREACHED_SLA. That overstates evidence. The classifier uses supplied elapsed numbers and a simple threshold; actual SLA policy may exclude weekends, pauses or particular ticket states. Names influence downstream decisions, so review output vocabulary with the same care as the boolean expression.

Worked laboratory solution

Compare your reasoning with the worked solution

The delete call violates report-only behavior, recursion exceeds the selected flat-directory scope, and >= includes the excluded boundary. Symlink treatment must be explicit and tested rather than assumed. A safer design returns candidate records from a pure classification function and keeps filesystem enumeration in a small reviewed adapter. Inject the reference time so tests are repeatable. Run only in a disposable fixture directory after reviewing the code; a test suite cannot undo an unsafe first execution.

Official reference for further study

Download the course workbook and practice materials