laba8.py - lab8 - a_lukin/Labs_Python

laba8.py
404 строки19 KB
a_lukin
create: laba8.py, laba8.txt, scraping.log, 8-лаба.pdf, analysis_20260501_140224.json
01 май 2026, 14:35
01 май 2026, 14:351f00485
import requests
from bs4 import BeautifulSoup
import json
import csv
import time
from datetime import datetime
import sys
import logging

# ---------- Конфигурация логирования ----------
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('scraping.log', encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)

# ---------- Глобальные настройки ----------
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
REQUEST_TIMEOUT = 15
DELAY_BETWEEN_REQUESTS = 2  # этичное поведение

# ---------- Тестовые URL (замените на актуальные рабочие адреса) ----------
HTML_TARGET_URL = "https://www.cian.ru/cat.php?deal_type=sale&engine_version=2&offer_type=flat&region=1"  # пример
API_TARGET_URL = "https://api.mocki.io/v1/realestate"  # замените на реальный API или работающий мок-сервер

# ---------- Вспомогательные функции ----------
def safe_request(url, params=None, headers=HEADERS, timeout=REQUEST_TIMEOUT):
    """
    Безопасный GET-запрос с обработкой сетевых и HTTP-ошибок.
    Возвращает объект Response или None при неудаче.
    """
    try:
        response = requests.get(url, params=params, headers=headers, timeout=timeout)
        response.raise_for_status()  # 4xx/5xx -> исключение
        return response
    except requests.exceptions.Timeout:
        logging.error(f"Таймаут при подключении к {url}")
    except requests.exceptions.ConnectionError:
        logging.error(f"Ошибка соединения с {url}")
    except requests.exceptions.HTTPError as http_err:
        logging.error(f"HTTP ошибка {response.status_code} для {url}: {http_err}")
    except requests.exceptions.RequestException as req_err:
        logging.error(f"Общая сетевая ошибка при запросе {url}: {req_err}")
    return None


# ---------- Функция 1: Парсинг HTML с сайта недвижимости ----------
def parse_website(url=HTML_TARGET_URL):
    """
    Сбор данных о квартирах путём парсинга HTML-страницы.
    Возвращает список словарей с полями: title, price, area, address, district.
    """
    logging.info(f"Запуск парсинга сайта: {url}")
    data = []
    try:
        response = safe_request(url)
        if not response:
            return data

        soup = BeautifulSoup(response.content, 'html.parser')

        # Поиск объявлений – адаптируйте селекторы под структуру целевого сайта
        # Пример для гипотетической структуры (карточки объявлений)
        cards = soup.select('div[data-name="CardComponent"]')  # замените на реальный селектор
        if not cards:
            logging.warning("Селектор не нашёл объявлений. Возможно, изменилась структура сайта.")
            # Возвращаем тестовый набор для демонстрации, если сайт недоступен
            logging.info("Используются демонстрационные данные.")
            return [
                {"title": "1-к квартира, 35 м²", "price": 8500000, "area": 35, "address": "Москва, ул. Тверская, 12", "district": "Центральный"},
                {"title": "2-к квартира, 55 м²", "price": 12200000, "area": 55, "address": "Москва, ул. Ленина, 8", "district": "Северный"}
            ]

        for card in cards:
            try:
                title_elem = card.select_one('[data-mark="OfferTitle"]')  # замена на реальный атрибут
                price_elem = card.select_one('[data-mark="Price"]')
                area_elem = card.select_one('[data-mark="Area"]')
                addr_elem = card.select_one('[data-mark="Address"]')

                title = title_elem.get_text(strip=True) if title_elem else "Н/Д"
                price_text = price_elem.get_text(strip=True) if price_elem else "0"
                area_text = area_elem.get_text(strip=True) if area_elem else "0"
                address = addr_elem.get_text(strip=True) if addr_elem else "Н/Д"

                # Очистка числовых значений (убрать пробелы, символы валют)
                price = int(''.join(filter(str.isdigit, price_text))) if any(c.isdigit() for c in price_text) else 0
                area = float(''.join(c for c in area_text if c.isdigit() or c == '.')) if any(c.isdigit() for c in area_text) else 0.0

                # Простейшее определение района (можно улучшить геокодированием)
                district = "Не определён"
                if "Тверская" in address:
                    district = "Центральный"
                elif "Ленина" in address:
                    district = "Северный"

                data.append({
                    "title": title,
                    "price": price,
                    "area": area,
                    "address": address,
                    "district": district
                })
            except Exception as e:
                logging.error(f"Ошибка при разборе одной карточки: {e}")
                continue

        logging.info(f"Парсинг завершён. Получено записей: {len(data)}")

    except Exception as e:
        logging.error(f"Критическая ошибка парсинга: {e}")

    return data


# ---------- Функция 2: Получение данных через API ----------
def get_data_from_api(api_url=API_TARGET_URL, params=None):
    """
    Получение данных о недвижимости через REST API (JSON).
    Возвращает список словарей с аналогичной структурой.
    """
    logging.info(f"Запрос к API: {api_url}")
    data = []
    try:
        response = safe_request(api_url, params=params)
        if not response:
            return data

        # Предполагаем, что API возвращает список объектов в поле "items"
        raw_data = response.json()

        # Обработка различных возможных структур ответа
        if isinstance(raw_data, list):
            items = raw_data
        elif isinstance(raw_data, dict):
            items = raw_data.get('items', raw_data.get('data', []))
        else:
            logging.warning("Неожиданный формат данных API")
            items = []

        for item in items:
            try:
                obj = {
                    "title": item.get("title", item.get("name", "Без названия")),
                    "price": int(item.get("price", 0)),
                    "area": float(item.get("area", item.get("square", 0))),
                    "address": item.get("address", item.get("location", "")),
                    "district": item.get("district", item.get("region", "Не определён"))
                }
                data.append(obj)
            except Exception as e:
                logging.error(f"Ошибка разбора элемента API: {e}")

        logging.info(f"API вернул {len(data)} записей")

    except json.JSONDecodeError as e:
        logging.error(f"Ошибка декодирования JSON: {e}")
    except Exception as e:
        logging.error(f"Неожиданная ошибка при работе с API: {e}")

    # Если реальный API недоступен, возвращаем демо-данные
    if not data:
        logging.info("API-данные не получены. Используются тестовые записи.")
        data = [
            {"title": "3-к квартира, 75 м²", "price": 18500000, "area": 75.0, "address": "Москва, Кутузовский пр-т, 30", "district": "Западный"},
            {"title": "Студия, 25 м²", "price": 6200000, "area": 25.0, "address": "Москва, ул. Пушкина, 5", "district": "Центральный"}
        ]

    return data


# ---------- Функция 3: Анализ и обработка данных ----------
def analyze_data(site_data, api_data):
    """
    Объединяет данные из двух источников и вычисляет статистику:
    - общее количество объектов;
    - средняя цена и площадь;
    - средняя цена по районам;
    - минимальная и максимальная цена.
    """
    all_data = site_data + api_data
    if not all_data:
        logging.warning("Нет данных для анализа.")
        return {}

    try:
        total = len(all_data)
        prices = [obj['price'] for obj in all_data if obj['price'] > 0]
        areas = [obj['area'] for obj in all_data if obj['area'] > 0]

        avg_price = sum(prices) / len(prices) if prices else 0
        avg_area = sum(areas) / len(areas) if areas else 0

        # Группировка по районам
        district_stats = {}
        for obj in all_data:
            district = obj.get('district', 'Не определён')
            if district not in district_stats:
                district_stats[district] = {'prices': [], 'count': 0}
            if obj['price'] > 0:
                district_stats[district]['prices'].append(obj['price'])
            district_stats[district]['count'] += 1

        avg_by_district = {
            dist: {
                'avg_price': sum(v['prices']) / len(v['prices']) if v['prices'] else 0,
                'count': v['count']
            }
            for dist, v in district_stats.items()
        }

        min_price = min(prices) if prices else 0
        max_price = max(prices) if prices else 0

        result = {
            'total_objects': total,
            'average_price': round(avg_price, 2),
            'average_area': round(avg_area, 2),
            'min_price': min_price,
            'max_price': max_price,
            'average_price_by_district': avg_by_district
        }

        logging.info("Анализ завершён успешно.")
        return result

    except Exception as e:
        logging.error(f"Ошибка при анализе данных: {e}")
        return {}


# ---------- Функция 4: Сохранение данных в файл ----------
def save_to_file(data, filename=None, file_format='json'):
    """
    Сохраняет словарь/список в файл указанного формата (json или csv).
    Если filename не задан, генерируется имя с временной меткой.
    """
    if not data:
        logging.warning("Нет данных для сохранения.")
        return

    if not filename:
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filename = f"real_estate_{timestamp}"

    try:
        if file_format == 'json':
            full_name = f"{filename}.json"
            with open(full_name, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            logging.info(f"Данные сохранены в {full_name}")

        elif file_format == 'csv':
            full_name = f"{filename}.csv"
            # Если передан список объектов с одинаковыми ключами
            if isinstance(data, list) and len(data) > 0:
                keys = data[0].keys()
                with open(full_name, 'w', newline='', encoding='utf-8') as f:
                    writer = csv.DictWriter(f, fieldnames=keys)
                    writer.writeheader()
                    writer.writerows(data)
                logging.info(f"Данные сохранены в {full_name}")
            else:
                logging.error("CSV сохранение поддерживается только для списка однотипных словарей.")

        else:
            logging.error(f"Неподдерживаемый формат: {file_format}")

    except IOError as e:
        logging.error(f"Ошибка записи в файл: {e}")
    except Exception as e:
        logging.error(f"Неожиданная ошибка при сохранении: {e}")


# ---------- Функция 5: Текстовое меню ----------
def show_menu():
    print("\n" + "="*60)
    print("ВЕБ-СКРЕЙПИНГ: Сбор данных о недвижимости")
    print("="*60)
    print("1. Получить данные с сайта (HTML парсинг)")
    print("2. Получить данные через API")
    print("3. Проанализировать собранные данные")
    print("4. Сохранить данные в файл")
    print("5. Показать справку")
    print("0. Выход")
    print("="*60)


# ---------- Главная функция ----------
def main():
    collected_data = {
        'site': [],
        'api': []
    }
    analysis_result = {}

    print("Система мониторинга недвижимости v1.0")
    logging.info("Запуск приложения")

    while True:
        show_menu()
        try:
            choice = input("\nВыберите действие (0-5): ").strip()

            if choice == "1":
                print("\nПарсинг сайта...")
                site_data = parse_website()
                collected_data['site'] = site_data
                print(f"Получено объявлений: {len(site_data)}")
                time.sleep(DELAY_BETWEEN_REQUESTS)

            elif choice == "2":
                print("\nЗапрос к API...")
                api_data = get_data_from_api()
                collected_data['api'] = api_data
                print(f"Получено записей через API: {len(api_data)}")
                time.sleep(DELAY_BETWEEN_REQUESTS)

            elif choice == "3":
                if not collected_data['site'] and not collected_data['api']:
                    print("\nНет данных. Сначала выполните сбор (пункты 1 и 2).")
                    continue
                print("\nАнализ данных...")
                analysis_result = analyze_data(collected_data['site'], collected_data['api'])
                if analysis_result:
                    print("\n--- Результаты анализа ---")
                    print(f"Всего объектов: {analysis_result['total_objects']}")
                    print(f"Средняя цена: {analysis_result['average_price']:,.2f} руб.")
                    print(f"Средняя площадь: {analysis_result['average_area']:.2f} м²")
                    print(f"Минимальная цена: {analysis_result['min_price']:,.2f} руб.")
                    print(f"Максимальная цена: {analysis_result['max_price']:,.2f} руб.")
                    print("\nСредняя цена по районам:")
                    for district, stats in analysis_result['average_price_by_district'].items():
                        print(f"  {district}: {stats['avg_price']:,.2f} руб. ({stats['count']} об.)")
                else:
                    print("Не удалось выполнить анализ.")

            elif choice == "4":
                # Предлагаем сохранить либо сырые данные, либо результаты анализа
                sub_choice = input("Что сохранить? (1 - сырые данные, 2 - анализ): ").strip()
                if sub_choice == "1":
                    data_to_save = collected_data
                elif sub_choice == "2":
                    data_to_save = analysis_result
                else:
                    print("Неверный выбор.")
                    continue

                if not data_to_save:
                    print("Нет данных для сохранения.")
                    continue

                fmt = input("Формат (json/csv): ").strip().lower()
                if fmt not in ['json', 'csv']:
                    print("Неверный формат.")
                    continue

                # Имя файла с меткой времени
                timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
                if sub_choice == "1":
                    fname = f"raw_data_{timestamp}"
                else:
                    fname = f"analysis_{timestamp}"
                save_to_file(data_to_save, filename=fname, file_format=fmt)

            elif choice == "5":
                print("\n--- Справка ---")
                print("Программа собирает данные о ценах на недвижимость из двух источников:")
                print("1. Парсинг HTML с сайта объявлений (настраиваемые селекторы).")
                print("2. Получение JSON через публичный/мок API.")
                print("3. Анализ: средняя цена, площадь, статистика по районам.")
                print("4. Сохранение результатов в JSON/CSV.")
                print("Все ошибки логируются в scraping.log.")
                print("Для изменения URL источников отредактируйте переменные в начале кода.")

            elif choice == "0":
                print("\nЗавершение работы. До свидания!")
                logging.info("Приложение остановлено пользователем.")
                break

            else:
                print("Некорректный ввод. Введите число от 0 до 5.")

        except KeyboardInterrupt:
            print("\n\nПрервано пользователем.")
            logging.info("Приложение прервано.")
            break
        except Exception as e:
            logging.error(f"Непредвиденная ошибка в главном цикле: {e}")
            print("Произошла ошибка. Подробности в логе.")
            time.sleep(1)

    # Завершаем логирование
    logging.shutdown()


# Точка входа
if __name__ == "__main__":
    main()