FreeCampus Python

Turn Messy Text into Trustworthy Records

Recognize and extract text patterns, validate structure and domain rules, and preserve accepted records, rejected evidence, and source lineage.
python-foundations files-paths-external-data regular-expressions validation
Open in Colab
  • Level: Python Foundations
  • Estimated time: 4–5 hours
  • You will learn: Prefer simple string operations when they fit, read a practical regular-expression subset, distinguish recognition from extraction, validate records in stages, choose fail-fast or error accumulation, and keep input lineage through a complete pipeline.
  • Practice in: Google Colab, JupyterLab, or a local editor with messy beacon records

Open this lesson in Google Colab

A radio observatory receives beacon lines written by several generations of equipment. A clean line looks like this:

BCN-2042 | Europa Relay | temperature=18.5; battery=87

Other lines have surrounding spaces, an identifier with extra characters, a missing station name, a nonnumeric temperature, or a battery outside 0–100. A single regular expression can recognize punctuation, but it should not be asked to make every domain decision. This lesson builds a sequence whose stages can be explained and tested independently.

1. Start with the simplest string operations that fit

For exact prefixes, suffixes, delimiters, or whitespace policy, ordinary string methods are usually clearer than a regular expression:

raw_line = "  BCN-2042 | Europa Relay | temperature=18.5; battery=87  "
line = raw_line.strip()

print(line.startswith("BCN-"))
print(line.count("|"))
print([part.strip() for part in line.split("|")])

This works if the contract is exactly three |-separated fields and the last field has a separately parsed structure. It becomes awkward if whitespace can vary within the punctuation or several identifier families are allowed.

Use a regular expression when you need a compact pattern language, not because it looks more advanced.

Need Clear first choice
exact prefix or suffix startswith, endswith
one literal delimiter split or partition
remove known surrounding whitespace strip under a stated policy
fixed character classes and repeated structure regular expression
a full structured interchange format its parser, not a regex

CSV quoting and JSON nesting are grammars with dedicated parsers. A regex should not replace them.

For a small, owned delimiter grammar, write the contract directly:

def split_beacon_fields(raw_line):
    """Return three stripped beacon fields from an exact pipe-delimited line."""
    parts = [part.strip() for part in raw_line.strip().split("|")]
    if len(parts) != 3:
        raise ValueError(f"expected three pipe-delimited fields; got {len(parts)}")
    beacon_id, station, readings = parts
    if not all([beacon_id, station, readings]):
        raise ValueError("beacon fields must be non-empty")
    return beacon_id, station, readings


assert split_beacon_fields(
    "BCN-2042 | Europa Relay | temperature=18.5; battery=87"
) == ("BCN-2042", "Europa Relay", "temperature=18.5; battery=87")

This parser is easier to change if the only syntax is three pipe-separated fields. The later regular expression is justified by the more detailed numeric grammar. Comparing the two implementations is a design exercise, not a contest to use the most punctuation.

2. Read a useful regular-expression subset

Python’s re module compiles a pattern and applies it to text. Raw string literals such as r"\d" keep backslashes from being interpreted first by the Python string-literal layer.

import re

identifier_pattern = re.compile(r"BCN-\d{4}")

for candidate in ["BCN-2042", "BCN-42", "xxBCN-2042yy"]:
    print(candidate, bool(identifier_pattern.fullmatch(candidate)))

Read BCN-\d{4} from left to right:

  • BCN- matches those literal characters;
  • \d matches a Unicode decimal digit by default; and
  • {4} requires exactly four repetitions of the preceding token.

If the external contract specifically means ASCII digits, express that as [0-9]{4} or use an ASCII flag. A pattern should encode the real contract, not a visually similar approximation.

Useful pieces for this unit:

Pattern piece Meaning Example
. almost any single character A.B
[A-Z] one character in the range uppercase ASCII letter
[^|] one character except | field content
\s whitespace character space, tab, or newline
? zero or one repetition optional sign
* zero or more repetitions possibly empty content
+ one or more repetitions non-empty content
{2,4} from two through four bounded repetition
(one|two) alternatives either literal choice
(...) capturing group extract by position
(?P<name>...) named group extract by meaning
^ and $ beginning and end anchors whole-line intent

Raw strings do not make a pattern automatically correct. Compile small pieces, try accepted and rejected examples, and explain each token.

3. Choose fullmatch, search, or extraction deliberately

Three operations answer different questions:

pattern = re.compile(r"BCN-[0-9]{4}")
text = "alert BCN-2042 received"

print("fullmatch:", pattern.fullmatch(text))
print("search:", pattern.search(text).group())
  • fullmatch asks whether the entire string satisfies the pattern;
  • search scans for a match anywhere; and
  • match starts at the beginning but can leave an unmatched suffix; and
  • finditer yields repeated matches with positions.

For a record identifier field, use fullmatch; accepting "xxBCN-2042yy" because it contains a valid-looking fragment violates the field contract. For scanning a log message that may mention an identifier, search is appropriate.

Named groups turn successful recognition into labeled extraction:

BEACON_PATTERN = re.compile(
    r"(?P<beacon_id>BCN-[0-9]{4})\s*\|\s*"
    r"(?P<station>[^|]+)\s*\|\s*"
    r"temperature=(?P<temperature>[+-]?[0-9]+(?:\.[0-9]+)?);\s*"
    r"battery=(?P<battery>[0-9]+)"
)

candidate = "BCN-2042 | Europa Relay | temperature=18.5; battery=87"
match = BEACON_PATTERN.fullmatch(candidate)
assert match is not None
print(match.groupdict())

The pattern recognizes syntax and extracts strings. It does not prove that battery 187 is reasonable, temperature is within sensor limits, or the station is registered.

A trustworthy pipeline keeps boundary responsibilities separate and records where a candidate leaves the accepted path.

flowchart LR
  raw["Raw text + source"] --> normalize["Normalize permitted whitespace"]
  normalize --> recognize["Recognize and extract"]
  recognize --> convert["Convert field types"]
  convert --> validate["Validate structure and domain"]
  validate -->|valid| accepted["Accepted + lineage"]
  validate -->|anticipated failure| rejected["Rejected + raw + reason"]

Unexpected programming exceptions leave the pipeline rather than being relabeled as bad input.

Checkpoint: use patterns for the right question

4. Normalize only what the contract allows

Normalization makes equivalent permitted inputs share one representation. It should be idempotent: applying it twice yields the same result as once.

def normalize_beacon_line(raw_line):
    """Remove surrounding whitespace while preserving internal field text."""
    return raw_line.strip()


sample = "  BCN-2042 | Europa Relay | temperature=18.5; battery=87  "
once = normalize_beacon_line(sample)
twice = normalize_beacon_line(once)
assert twice == once

Do not automatically collapse all internal whitespace, lowercase display names, or delete punctuation. Those transformations may change meaning or destroy forensic evidence. Preserve the original alongside any comparison form.

A record can carry lineage explicitly:

candidate = {
    "source": "receiver-europa.log",
    "record_number": 17,
    "raw": sample,
    "normalized": normalize_beacon_line(sample),
}

Once data has passed through several functions, the source and record number are still available for explanations.

5. Convert values with chained context

After a successful match, convert fields under narrow exception boundaries:

def convert_beacon_groups(groups, source, record_number):
    """Convert extracted beacon strings into typed values."""
    try:
        temperature = float(groups["temperature"])
    except ValueError as error:
        raise ValueError(
            f"{source}: record {record_number}: invalid temperature "
            f"{groups['temperature']!r}"
        ) from error

    try:
        battery = int(groups["battery"])
    except ValueError as error:
        raise ValueError(
            f"{source}: record {record_number}: invalid battery "
            f"{groups['battery']!r}"
        ) from error

    return {
        "beacon_id": groups["beacon_id"],
        "station": groups["station"].strip(),
        "temperature": temperature,
        "battery": battery,
    }

The regex already constrains these particular numeric strings, but keeping conversion checks close to the conversion makes the function robust if it is later reused with groups from another source. Exception chaining preserves the original failure.

6. Structural, field, and cross-field rules are distinct

A useful validator checks increasingly domain-specific claims:

KNOWN_STATIONS = {"Europa Relay", "Luna South", "Mars Ridge"}


def validate_beacon(record):
    """Return record after checking field and cross-field rules."""
    required = {"beacon_id", "station", "temperature", "battery"}
    missing = required - record.keys()
    extra = record.keys() - required
    if missing:
        raise ValueError(f"missing fields: {sorted(missing)!r}")
    if extra:
        raise ValueError(f"unexpected fields: {sorted(extra)!r}")

    if not record["station"]:
        raise ValueError("station must be non-empty")
    if record["station"] not in KNOWN_STATIONS:
        raise ValueError(f"unknown station: {record['station']!r}")
    if not -120 <= record["temperature"] <= 80:
        raise ValueError("temperature outside sensor range -120..80")
    if not 0 <= record["battery"] <= 100:
        raise ValueError("battery must be between 0 and 100")
    if record["temperature"] < -80 and record["battery"] < 10:
        raise ValueError("extreme cold reports require battery of at least 10")

    return record

The sets of keys are structural rules. Station membership, ranges, and the relationship between cold temperature and battery are domain rules. A program can accumulate several messages for a form-like user experience, or fail at the first violation for a simple import API.

Some rules apply across records rather than inside one record. Duplicate IDs need batch state:

def require_unique_beacon_ids(records):
    """Return records after rejecting the first duplicated beacon identifier."""
    seen = set()
    for record in records:
        beacon_id = record["beacon_id"]
        if beacon_id in seen:
            raise ValueError(f"duplicate beacon_id: {beacon_id!r}")
        seen.add(beacon_id)
    return records


unique_preview = [
    {"beacon_id": "BCN-1001"},
    {"beacon_id": "BCN-1002"},
]
assert require_unique_beacon_ids(unique_preview) is unique_preview

Choose whether a duplicate rejects only the later record, both records, or the whole batch. That decision requires source order and ownership policy; a single-record regex cannot make it.

Choose and document one style:

def collect_battery_errors(value):
    """Return all independently checkable battery validation messages."""
    errors = []
    if not isinstance(value, int) or isinstance(value, bool):
        errors.append("battery must be an integer")
        return errors
    if value < 0:
        errors.append("battery cannot be negative")
    if value > 100:
        errors.append("battery cannot exceed 100")
    return errors

The early return avoids meaningless numeric comparisons after a type failure. Do not continue merely to produce more messages when later rules require a valid earlier stage.

Use a rule table before writing a large validator. It exposes dependencies and prevents an error message from claiming more than has been checked:

Order Rule Depends on Failure category
1 value is a mapping with exact keys parsed value structure
2 identifier, station, and numeric fields have expected basic types structure field type
3 identifier follows the complete pattern text identifier recognition
4 station is known and ranges are satisfied converted fields domain
5 extreme cold has sufficient battery valid temperature and battery cross-field
6 identifier has not appeared earlier accepted batch state cross-record

Independent field rules can accumulate messages. Dependent rules should wait. For example, after battery fails the integer type check, “battery exceeds 100” is not a second useful diagnosis; comparison is not yet meaningful. That is why collect_battery_errors returns immediately after the type failure.

Error accumulation also changes the return contract. A function that raises one ValueError cannot simultaneously return a list of all field errors. Choose a consistent API such as validate_or_raise(record) for fail-fast code or validation_messages(record) returning a list for batch feedback, and name it so callers know which behavior to expect.

Checkpoint: keep validation stages explainable

7. Compose the pipeline and quarantine expected rejections

One function can connect the stages without blending their responsibilities:

def parse_beacon_line(raw_line, source, record_number):
    """Return one validated beacon record or raise contextual ValueError."""
    normalized = normalize_beacon_line(raw_line)
    match = BEACON_PATTERN.fullmatch(normalized)
    if match is None:
        raise ValueError(
            f"{source}: record {record_number}: line does not match beacon syntax"
        )

    converted = convert_beacon_groups(match.groupdict(), source, record_number)
    return validate_beacon(converted)

A batch loader can quarantine anticipated ValueError instances:

def partition_beacon_lines(lines, source):
    """Return accepted records and detailed anticipated rejections."""
    accepted = []
    rejected = []

    for record_number, raw_line in enumerate(lines, start=1):
        try:
            record = parse_beacon_line(raw_line, source, record_number)
        except ValueError as error:
            rejected.append(
                {
                    "source": source,
                    "record": record_number,
                    "raw": raw_line,
                    "reason": str(error),
                }
            )
        else:
            accepted.append(
                {
                    **record,
                    "source": source,
                    "record": record_number,
                }
            )

    return accepted, rejected

Why not catch Exception? Examine a plausible defect:

def transform_beacon(record):
    """Return a display label for one parsed record."""
    return f"{record['beacon_id']} from {record['station']}"


try:
    transform_beacon(None)
except TypeError as error:
    print("programming failure remains visible:", error)

If a batch loop catches Exception and labels this as “invalid source data,” it hides a programming defect. Catch the exception type your parser deliberately uses for anticipated record rejection. Let TypeError, KeyError from an internal typo, and other unexpected defects stop the run with their traceback.

8. Decode the observatory beacon batch

Use these inputs:

beacon_lines = [
    "BCN-2042 | Europa Relay | temperature=18.5; battery=87",
    "xxBCN-2043yy | Luna South | temperature=20; battery=90",
    "BCN-2044 | Unknown Port | temperature=12; battery=70",
    "BCN-2045 | Mars Ridge | temperature=-95; battery=8",
    "BCN-2046 | Luna South | temperature=21; battery=101",
    "  BCN-2047 | Mars Ridge | temperature=-20; battery=65  ",
]

Write this prediction table before calling the pipeline:

Record Expected stage Expected result
1 all stages accept BCN-2042
2 full-line recognition reject wrapper text around the identifier
3 domain validation reject the unknown station
4 cross-field validation reject extreme cold with battery below 10
5 field range reject battery 101
6 normalization then all stages accept BCN-2047

For one accepted line, also run split_beacon_fields. Compare its three broad fields with the regex named groups. The split version is easier to read for the outer pipes; the regex earns its complexity by recognizing and extracting the numeric readings. A production parser could combine both rather than force one tool to express every layer.

Before running, predict which records are accepted and which stage rejects each other record. Then:

  1. call partition_beacon_lines with source receiver-night.log;
  2. assert accepted IDs are exactly BCN-2042 and BCN-2047;
  3. assert every rejection retains source, record, raw line, and non-empty reason;
  4. assert normalize_beacon_line is idempotent for every input;
  5. temporarily change validate_beacon to call a misspelled helper and confirm the resulting NameError is not quarantined as bad data;
  6. restore the function and rerun from a clean state.
Check the expected partition and evidence after your clean run 
accepted, rejected = partition_beacon_lines(
    beacon_lines,
    "receiver-night.log",
)

assert [record["beacon_id"] for record in accepted] == ["BCN-2042", "BCN-2047"]
assert len(rejected) == 4
assert all(item["source"] == "receiver-night.log" for item in rejected)
assert all(isinstance(item["record"], int) for item in rejected)
assert all(item["raw"] in beacon_lines for item in rejected)
assert all(item["reason"] for item in rejected)
assert all(
    normalize_beacon_line(normalize_beacon_line(line))
    == normalize_beacon_line(line)
    for line in beacon_lines
)

Inspect the reasons instead of only counting them. One should describe syntax, one an unknown station, one the cross-field extreme-cold rule, and one battery range. That diversity proves several stages were exercised.

Checkpoint: preserve expected and unexpected failures

9. Key points for trustworthy validation

  • Prefer clear string methods for simple delimiters and exact prefixes; use a regular expression when a pattern language genuinely clarifies the contract.
  • fullmatch validates a whole field, while search finds a substring. Choose according to the question you are asking.
  • Recognition, extraction, conversion, structural validation, field rules, and cross-field rules are separate responsibilities.
  • Preserve original values and lineage. Make permitted normalization idempotent instead of gradually changing data on reruns.
  • Choose fail-fast or accumulated errors based on how later rules depend on earlier success.
  • Quarantine anticipated invalid records with useful evidence; do not convert unexpected programming failures into ordinary rejections.

10. References and next step

Next, move below the text decoder. You will inspect raw byte values, file signatures, binary chunks, mutable byte buffers, and in-memory streams without pretending arbitrary binary data is encoded prose.

Back to top