| """CLI interface for address conversion.""" |
|
|
| import csv |
| import sys |
|
|
| import click |
|
|
| from .converter import convert_address, batch_convert |
|
|
|
|
| @click.group() |
| def main(): |
| """Vietnamese address converter (post-merger 01/07/2025).""" |
| pass |
|
|
|
|
| @main.command() |
| @click.argument("address") |
| def convert(address): |
| """Convert a single address. |
| |
| Example: address-convert convert "Phường Phúc Xá, Quận Ba Đình, Thành phố Hà Nội" |
| """ |
| result = convert_address(address) |
| click.echo(f"Input: {result.original}") |
| click.echo(f"Output: {result.converted}") |
| click.echo(f"Status: {result.status.value}") |
| if result.mapping_type: |
| click.echo(f"Type: {result.mapping_type.value}") |
| if result.note: |
| click.echo(f"Note: {result.note}") |
|
|
|
|
| @main.command() |
| @click.argument("input_file", type=click.Path(exists=True)) |
| @click.argument("output_file", type=click.Path()) |
| @click.option("--column", "-c", default="address", help="Column name containing addresses") |
| def batch(input_file, output_file, column): |
| """Convert addresses from a CSV file. |
| |
| Reads INPUT_FILE CSV, converts the address column, writes to OUTPUT_FILE. |
| """ |
| with open(input_file, newline="", encoding="utf-8") as f: |
| reader = csv.DictReader(f) |
| if column not in reader.fieldnames: |
| click.echo(f"Error: Column '{column}' not found. Available: {reader.fieldnames}", err=True) |
| sys.exit(1) |
|
|
| addresses = [] |
| rows = [] |
| for row in reader: |
| rows.append(row) |
| addresses.append(row[column]) |
|
|
| results = batch_convert(addresses) |
|
|
| fieldnames = list(rows[0].keys()) + ["converted_address", "conversion_status", "mapping_type"] |
| with open(output_file, "w", newline="", encoding="utf-8") as f: |
| writer = csv.DictWriter(f, fieldnames=fieldnames) |
| writer.writeheader() |
| for row, result in zip(rows, results): |
| row["converted_address"] = result.converted |
| row["conversion_status"] = result.status.value |
| row["mapping_type"] = result.mapping_type.value if result.mapping_type else "" |
| writer.writerow(row) |
|
|
| |
| total = len(results) |
| success = sum(1 for r in results if r.status.value == "success") |
| partial = sum(1 for r in results if r.status.value == "partial") |
| not_found = sum(1 for r in results if r.status.value == "not_found") |
| click.echo(f"Converted {total} addresses: {success} success, {partial} partial, {not_found} not found") |
| click.echo(f"Output: {output_file}") |
|
|
