client_manager_simple.py - master - Askoline/Homework_DB2

client_manager_simple.py
300 строк11 KB
Askoline
Initial commit: система управления клиентами
02 авг 2026, 14:28
02 авг 2026, 14:281333054
"""
Упрощенная программа управления клиентами
"""
import psycopg2
from psycopg2 import sql

# КОНФИГУРАЦИЯ БД - ИЗМЕНИТЕ ПАРОЛЬ ЕСЛИ НУЖНО
DB_CONFIG = {
    'database': 'clients_db',
    'user': 'postgres',
    'password': 'postgres',  # Если пароль другой, измените здесь
    'host': 'localhost',
    'port': '5432'
}

def create_db(conn):
    """Создание таблиц"""
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS clients (
                id SERIAL PRIMARY KEY,
                first_name VARCHAR(50) NOT NULL,
                last_name VARCHAR(50) NOT NULL,
                email VARCHAR(100) UNIQUE NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS phones (
                id SERIAL PRIMARY KEY,
                client_id INTEGER NOT NULL REFERENCES clients(id) ON DELETE CASCADE,
                phone_number VARCHAR(20) NOT NULL,
                UNIQUE(client_id, phone_number)
            )
        """)
        conn.commit()
        print("✓ Таблицы созданы")

def add_client(conn, first_name, last_name, email, phones=None):
    """Добавление клиента"""
    with conn.cursor() as cur:
        try:
            cur.execute("""
                INSERT INTO clients (first_name, last_name, email)
                VALUES (%s, %s, %s)
                RETURNING id
            """, (first_name, last_name, email))
            client_id = cur.fetchone()[0]
            
            if phones:
                for phone in phones:
                    add_phone(conn, client_id, phone)
            
            conn.commit()
            print(f"✓ Клиент '{first_name} {last_name}' добавлен (ID: {client_id})")
            return client_id
        except psycopg2.IntegrityError as e:
            conn.rollback()
            raise ValueError(f"Email '{email}' уже существует")

def add_phone(conn, client_id, phone):
    """Добавление телефона"""
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
        if not cur.fetchone():
            raise ValueError(f"Клиент с ID {client_id} не найден")
        
        try:
            cur.execute("""
                INSERT INTO phones (client_id, phone_number)
                VALUES (%s, %s)
            """, (client_id, phone))
            conn.commit()
            print(f"✓ Телефон '{phone}' добавлен")
        except psycopg2.IntegrityError:
            conn.rollback()
            raise ValueError(f"Телефон '{phone}' уже существует")

def change_client(conn, client_id, first_name=None, last_name=None, email=None):
    """Изменение данных клиента"""
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
        if not cur.fetchone():
            raise ValueError(f"Клиент с ID {client_id} не найден")
        
        updates = []
        params = []
        if first_name:
            updates.append("first_name = %s")
            params.append(first_name)
        if last_name:
            updates.append("last_name = %s")
            params.append(last_name)
        if email:
            updates.append("email = %s")
            params.append(email)
        
        if not updates:
            return
        
        params.append(client_id)
        query = sql.SQL("UPDATE clients SET {} WHERE id = %s").format(
            sql.SQL(", ").join(map(sql.SQL, updates))
        )
        
        try:
            cur.execute(query, params)
            conn.commit()
            print(f"✓ Данные клиента {client_id} обновлены")
        except psycopg2.IntegrityError:
            conn.rollback()
            raise ValueError(f"Email '{email}' уже используется")

def delete_phone(conn, client_id, phone):
    """Удаление телефона"""
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
        if not cur.fetchone():
            raise ValueError(f"Клиент с ID {client_id} не найден")
        
        cur.execute("""
            DELETE FROM phones 
            WHERE client_id = %s AND phone_number = %s
        """, (client_id, phone))
        
        if cur.rowcount == 0:
            raise ValueError(f"Телефон '{phone}' не найден")
        
        conn.commit()
        print(f"✓ Телефон '{phone}' удален")

def delete_client(conn, client_id):
    """Удаление клиента"""
    with conn.cursor() as cur:
        cur.execute("SELECT id FROM clients WHERE id = %s", (client_id,))
        if not cur.fetchone():
            raise ValueError(f"Клиент с ID {client_id} не найден")
        
        cur.execute("DELETE FROM clients WHERE id = %s", (client_id,))
        conn.commit()
        print(f"✓ Клиент {client_id} удален")

def find_client(conn, first_name=None, last_name=None, email=None, phone=None):
    """Поиск клиента"""
    with conn.cursor() as cur:
        conditions = []
        params = []
        
        if first_name:
            conditions.append("c.first_name ILIKE %s")
            params.append(f"%{first_name}%")
        if last_name:
            conditions.append("c.last_name ILIKE %s")
            params.append(f"%{last_name}%")
        if email:
            conditions.append("c.email ILIKE %s")
            params.append(f"%{email}%")
        if phone:
            conditions.append("p.phone_number ILIKE %s")
            params.append(f"%{phone}%")
        
        if not conditions:
            return []
        
        query = """
            SELECT DISTINCT c.id, c.first_name, c.last_name, c.email
            FROM clients c
            LEFT JOIN phones p ON c.id = p.client_id
            WHERE {}
            ORDER BY c.last_name, c.first_name
        """.format(" OR ".join(conditions))
        
        cur.execute(query, params)
        results = cur.fetchall()
        
        clients = []
        for row in results:
            cur.execute("SELECT phone_number FROM phones WHERE client_id = %s", (row[0],))
            phones = [p[0] for p in cur.fetchall()]
            clients.append({
                'id': row[0],
                'first_name': row[1],
                'last_name': row[2],
                'email': row[3],
                'phones': phones
            })
        return clients

def get_all_clients(conn):
    """Получение всех клиентов"""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, first_name, last_name, email, created_at
            FROM clients
            ORDER BY last_name, first_name
        """)
        clients = []
        for row in cur.fetchall():
            cur.execute("SELECT phone_number FROM phones WHERE client_id = %s", (row[0],))
            phones = [p[0] for p in cur.fetchall()]
            clients.append({
                'id': row[0],
                'first_name': row[1],
                'last_name': row[2],
                'email': row[3],
                'phones': phones
            })
        return clients

def print_client(client):
    """Вывод клиента"""
    phones = ", ".join(client['phones']) if client['phones'] else "нет телефонов"
    print(f"  {client['first_name']} {client['last_name']} (ID: {client['id']})")
    print(f"    Email: {client['email']}")
    print(f"    Телефоны: {phones}")
    print()

def demo():
    """Демонстрация работы"""
    print("=" * 60)
    print("СИСТЕМА УПРАВЛЕНИЯ КЛИЕНТАМИ")
    print("=" * 60)
    
    # Проверяем подключение
    try:
        conn = psycopg2.connect(**DB_CONFIG)
        print("✓ Подключено к PostgreSQL")
    except Exception as e:
        print(f"✗ Ошибка подключения: {e}")
        print("\nПроверьте параметры в DB_CONFIG:")
        print("  - Запущен ли PostgreSQL?")
        print("  - Правильные ли user и password?")
        print("\nПопробуйте:")
        print("  brew services start postgresql")
        return
    
    try:
        print("\n1. СОЗДАНИЕ ТАБЛИЦ")
        print("-" * 40)
        create_db(conn)
        
        print("\n2. ДОБАВЛЕНИЕ КЛИЕНТОВ")
        print("-" * 40)
        id1 = add_client(conn, "Иван", "Петров", "ivan@example.com", 
                        ["+7-999-123-45-67", "+7-999-987-65-43"])
        id2 = add_client(conn, "Мария", "Сидорова", "maria@example.com", 
                        ["+7-999-111-22-33"])
        id3 = add_client(conn, "Алексей", "Иванов", "alex@example.com")
        
        print("\n3. ДОБАВЛЕНИЕ ТЕЛЕФОНА")
        print("-" * 40)
        add_phone(conn, id1, "+7-999-555-44-33")
        
        print("\n4. ИЗМЕНЕНИЕ ДАННЫХ")
        print("-" * 40)
        change_client(conn, id2, last_name="Смирнова")
        
        print("\n5. УДАЛЕНИЕ ТЕЛЕФОНА")
        print("-" * 40)
        delete_phone(conn, id1, "+7-999-987-65-43")
        
        print("\n6. ПОИСК")
        print("-" * 40)
        
        print("\n  Поиск по имени 'Иван':")
        for c in find_client(conn, first_name="Иван"):
            print_client(c)
        
        print("\n  Поиск по телефону '+7-999-111':")
        for c in find_client(conn, phone="+7-999-111"):
            print_client(c)
        
        print("\n  Все клиенты:")
        for c in get_all_clients(conn):
            print_client(c)
        
        print("\n7. УДАЛЕНИЕ КЛИЕНТА")
        print("-" * 40)
        delete_client(conn, id3)
        
        print("\n8. ОШИБКИ")
        print("-" * 40)
        try:
            add_client(conn, "Дубликат", "Тестов", "ivan@example.com")
        except ValueError as e:
            print(f"  ✗ {e}")
        
        print("\n" + "=" * 60)
        print("ДЕМОНСТРАЦИЯ ЗАВЕРШЕНА")
        print("=" * 60)
        
    except Exception as e:
        print(f"✗ Ошибка: {e}")
        conn.rollback()
    finally:
        conn.close()
        print("\n✓ Соединение закрыто")

if __name__ == "__main__":
    demo()