34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Inspect customers/opportunities involved in a duplicate NIF conflict."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.commercial_service import normalize_tax_id
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--tax-id", required=True)
|
|
args = parser.parse_args()
|
|
tax_id = normalize_tax_id(args.tax_id)
|
|
with engine.begin() as conn:
|
|
customers = conn.execute(text("""
|
|
SELECT c.id::text, c.name, c.tax_id, c.email, c.phone, c.street_name, c.postal_zone, c.city_name,
|
|
c.country, c.created_at, c.updated_at,
|
|
(SELECT count(*) FROM opportunities o WHERE o.local_customer_id = c.id)::int AS opportunities,
|
|
(SELECT count(*) FROM commercial_documents cd WHERE cd.customer_id = c.id)::int AS documents
|
|
FROM customers c
|
|
WHERE c.tax_id = :tax_id OR c.name ILIKE '%' || :tax_id || '%'
|
|
ORDER BY c.tax_id NULLS LAST, c.updated_at DESC
|
|
"""), {"tax_id": tax_id}).mappings().all()
|
|
print(json.dumps({"tax_id": tax_id, "customers": [dict(c) for c in customers]}, ensure_ascii=False, indent=2, default=str))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|