|
|
"""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}") |
|
|
|
