src/price_lookup.py

"""實價登錄查詢模組 - Real Estate Price Lookup Module.

This module provides functionality to query Taiwan's real estate
transaction price registration data (實價登錄) from government open data.

Data source: 內政部不動產交易實價查詢服務網
https://lvr.land.moi.gov.tw/
"""

import logging
from dataclasses import dataclass
from typing import Any, Optional

import pandas as pd
import requests

import random

logger = logging.getLogger(__name__)

# 內政部實價登錄 API 端點
# 注意：這是示範用的模擬數據，實際使用需要申請 API 金鑰
BASE_URL = "https://lvr.land.moi.gov.tw/SERVICE/QueryPrice/QueryPrice"

# 台北市各區的區域代碼
TAIPEI_DISTRICTS = {
    "中正區": "A01",
    "大同區": "A02",
    "中山區": "A03",
    "松山區": "A04",
    "大安區": "A05",
    "萬華區": "A06",
    "信義區": "A07",
    "士林區": "A08",
    "北投區": "A09",
    "內湖區": "A10",
    "南港區": "A11",
    "文山區": "A12",
}

# 新北市各區
NEW_TAIPEI_DISTRICTS = {
    "板橋區": "F01",
    "三重區": "F02",
    "中和區": "F03",
    "永和區": "F04",
    "新莊區": "F05",
    "新店區": "F06",
    "土城區": "F07",
    "蘆洲區": "F08",
    "汐止區": "F09",
    "樹林區": "F10",
}


@dataclass
class PriceRecord:
    """Real estate transaction price record."""

    transaction_date: str
    district: str
    address: str
    property_type: str  # 土地/建物/車位
    area_ping: float
    total_price: float
    unit_price: float
    floor: Optional[str] = None
    building_age: Optional[int] = None

def _generate_mock_records(district: str, item_type: str) -> list[PriceRecord]:
    """Generate synthetic mock records for districts without sample data."""
    # Seed by district and type for consistent results
    seed_val = sum(ord(c) for c in district + item_type)
    rng = random.Random(seed_val)
    
    records = []
    
    # Base prices per type (arbitrary baseline)
    base_price_map = {
        "土地": (150, 400),  # Unit: 10k/ping
        "建物": (60, 150),   # Unit: 10k/ping
        "車位": (180, 350),  # Unit: 10k/space
    }
    
    # Streets pool
    streets = ["中正路", "中山路", "中華路", "信義路", "和平路", "仁愛路", 
               "建國路", "復興路", "忠孝路", "民權路"]
               
    min_p, max_p = base_price_map.get(item_type, (50, 100))
    
    # Generate 3-5 records
    count = rng.randint(3, 5)
    for i in range(count):
        month = rng.randint(10, 12)
        street = rng.choice(streets)
        section = rng.randint(1, 5)
        address = f"{district}{street}{section}段"
        
        if item_type == "土地":
            area = rng.randint(30, 200)
            unit_price = rng.randint(min_p, max_p)
            total_price = area * unit_price
            
            records.append(PriceRecord(
                transaction_date=f"2024-{month}",
                district=district,
                address=address,
                property_type="土地",
                area_ping=float(area),
                total_price=float(total_price),
                unit_price=float(unit_price)
            ))
            
        elif item_type == "建物":
            area = rng.randint(20, 80)
            unit_price = rng.randint(min_p, max_p)
            total_price = area * unit_price
            floor = rng.randint(2, 20)
            total_floors = floor + rng.randint(2, 10)
            age = rng.randint(0, 40)
            
            records.append(PriceRecord(
                transaction_date=f"2024-{month}",
                district=district,
                address=address,
                property_type="建物",
                area_ping=float(area),
                total_price=float(total_price),
                unit_price=float(unit_price),
                floor=f"{floor}F/{total_floors}F",
                building_age=age
            ))
            
        elif item_type == "車位":
            area = rng.randint(8, 12)
            total_price = rng.randint(min_p, max_p)
            unit_price = total_price / area
            
            records.append(PriceRecord(
                transaction_date=f"2024-{month}",
                district=district,
                address=address,
                property_type="車位",
                area_ping=float(area),
                total_price=float(total_price),
                unit_price=float(unit_price)
            ))
            
    return records


def get_sample_land_prices(district: str) -> list[PriceRecord]:
    """Get sample land transaction prices for a district.

    Args:
        district: District name (e.g., "大安區").

    Returns:
        list: List of PriceRecord objects.

    Note:
        This uses simulated data. For production, integrate with actual API.
    """
    # 模擬實價登錄資料 - 實際使用時需接入政府 API
    sample_data = {
        "大安區": [
            PriceRecord("2024-12", "大安區", "復興南路二段", "土地", 50.0, 15000, 300),
            PriceRecord("2024-11", "大安區", "敦化南路一段", "土地", 80.0, 28000, 350),
            PriceRecord("2024-10", "大安區", "仁愛路四段", "土地", 120.0, 48000, 400),
        ],
        "信義區": [
            PriceRecord("2024-12", "信義區", "松仁路", "土地", 60.0, 21000, 350),
            PriceRecord("2024-11", "信義區", "忠孝東路五段", "土地", 45.0, 15750, 350),
            PriceRecord("2024-10", "信義區", "基隆路一段", "土地", 100.0, 32000, 320),
        ],
        "中山區": [
            PriceRecord("2024-12", "中山區", "南京東路三段", "土地", 55.0, 13750, 250),
            PriceRecord("2024-11", "中山區", "民生東路三段", "土地", 70.0, 17500, 250),
        ],
    }
    
    if district in sample_data:
        return sample_data[district]
        
    return _generate_mock_records(district, "土地")


def get_sample_building_prices(district: str) -> list[PriceRecord]:
    """Get sample building transaction prices for a district.

    Args:
        district: District name.

    Returns:
        list: List of PriceRecord objects.
    """
    sample_data = {
        "大安區": [
            PriceRecord(
                "2024-12", "大安區", "復興南路二段", "建物",
                35.0, 3500, 100, "5F/12F", 15
            ),
            PriceRecord(
                "2024-11", "大安區", "敦化南路一段", "建物",
                45.0, 5850, 130, "8F/20F", 5
            ),
            PriceRecord(
                "2024-10", "大安區", "仁愛路四段", "建物",
                60.0, 9000, 150, "12F/25F", 3
            ),
        ],
        "信義區": [
            PriceRecord(
                "2024-12", "信義區", "松仁路", "建物",
                40.0, 5200, 130, "15F/30F", 2
            ),
            PriceRecord(
                "2024-11", "信義區", "忠孝東路五段", "建物",
                50.0, 6000, 120, "10F/22F", 8
            ),
        ],
        "中山區": [
            PriceRecord(
                "2024-12", "中山區", "南京東路三段", "建物",
                30.0, 2400, 80, "6F/14F", 20
            ),
            PriceRecord(
                "2024-11", "中山區", "民生東路三段", "建物",
                42.0, 3360, 80, "9F/18F", 12
            ),
        ],
    }

    
    if district in sample_data:
        return sample_data[district]
        
    return _generate_mock_records(district, "建物")


def get_sample_parking_prices(district: str) -> list[PriceRecord]:
    """Get sample parking space transaction prices for a district.

    Args:
        district: District name.

    Returns:
        list: List of PriceRecord objects.
    """
    sample_data = {
        "大安區": [
            PriceRecord("2024-12", "大安區", "復興南路二段", "車位", 8.0, 280, 35),
            PriceRecord("2024-11", "大安區", "敦化南路一段", "車位", 10.0, 380, 38),
            PriceRecord("2024-10", "大安區", "仁愛路四段", "車位", 12.0, 480, 40),
        ],
        "信義區": [
            PriceRecord("2024-12", "信義區", "松仁路", "車位", 9.0, 360, 40),
            PriceRecord("2024-11", "信義區", "忠孝東路五段", "車位", 8.5, 340, 40),
        ],
        "中山區": [
            PriceRecord("2024-12", "中山區", "南京東路三段", "車位", 8.0, 200, 25),
            PriceRecord("2024-11", "中山區", "民生東路三段", "車位", 9.0, 225, 25),
        ],
    }
    
    if district in sample_data:
        return sample_data[district]
        
    return _generate_mock_records(district, "車位")


def query_real_prices(
    city: str,
    district: str,
    property_type: str,
) -> pd.DataFrame:
    """Query real estate transaction prices.

    Args:
        city: City name (e.g., "台北市").
        district: District name (e.g., "大安區").
        property_type: Type of property ("土地", "建物", "車位").

    Returns:
        pd.DataFrame: Transaction records as DataFrame.
    """
    # 根據類型獲取樣本數據
    if property_type == "土地":
        records = get_sample_land_prices(district)
    elif property_type == "建物":
        records = get_sample_building_prices(district)
    elif property_type == "車位":
        records = get_sample_parking_prices(district)
    else:
        records = []

    if not records:
        return pd.DataFrame()

    # 轉換為 DataFrame
    data = []
    for r in records:
        row = {
            "交易日期": r.transaction_date,
            "區域": r.district,
            "地址": r.address,
            "類型": r.property_type,
            "面積(坪)": r.area_ping,
            "總價(萬)": r.total_price,
            "單價(萬/坪)": r.unit_price,
        }
        if r.floor:
            row["樓層"] = r.floor
        if r.building_age:
            row["屋齡"] = r.building_age
        data.append(row)

    return pd.DataFrame(data)


def get_average_prices(city: str, district: str) -> dict[str, float]:
    """Get average prices by property type for a district.

    Args:
        city: City name.
        district: District name.

    Returns:
        dict: Average unit prices by property type.
    """
    result = {}

    # 土地平均單價
    land_df = query_real_prices(city, district, "土地")
    if not land_df.empty:
        result["土地平均單價"] = land_df["單價(萬/坪)"].mean()

    # 建物平均單價
    building_df = query_real_prices(city, district, "建物")
    if not building_df.empty:
        result["建物平均單價"] = building_df["單價(萬/坪)"].mean()

    # 車位平均總價
    parking_df = query_real_prices(city, district, "車位")
    if not parking_df.empty:
        result["車位平均總價"] = parking_df["總價(萬)"].mean()

    return result


def fetch_actual_prices_from_api(
    city_code: str,
    district_code: str,
    year: int,
    season: int,
) -> Optional[pd.DataFrame]:
    """Fetch actual transaction data from government API.

    Args:
        city_code: City code (e.g., "A" for 台北市).
        district_code: District code.
        year: Year in ROC calendar (e.g., 113).
        season: Season (1-4).

    Returns:
        pd.DataFrame or None: Transaction data if successful.

    Note:
        This requires actual API access. The open data can be downloaded from:
        https://plvr.land.moi.gov.tw/DownloadOpenData
    """
    # 實價登錄開放資料下載位置
    # 格式：https://plvr.land.moi.gov.tw/DownloadSeason?season={year}S{season}&type=zip&fileName=lvr_landcsv.zip
    download_url = (
        f"https://plvr.land.moi.gov.tw/DownloadSeason?"
        f"season={year}S{season}&type=zip&fileName=lvr_landcsv.zip"
    )

    try:
        # 這裡示範如何下載，實際實作需要解壓縮和解析 CSV
        logger.info(f"Would fetch from: {download_url}")
        return None
    except requests.RequestException as e:
        logger.error(f"Failed to fetch price data: {e}")
        return None


def get_district_statistics(district: str) -> dict[str, Any]:
    """Get comprehensive statistics for a district.

    Args:
        district: District name.

    Returns:
        dict: Statistics including price trends and transaction counts.
    """
    land_prices = get_sample_land_prices(district)
    building_prices = get_sample_building_prices(district)
    parking_prices = get_sample_parking_prices(district)

    return {
        "district": district,
        "land_transactions": len(land_prices),
        "building_transactions": len(building_prices),
        "parking_transactions": len(parking_prices),
        "land_avg_price": (
            sum(p.unit_price for p in land_prices) / len(land_prices)
            if land_prices else 0
        ),
        "building_avg_price": (
            sum(p.unit_price for p in building_prices) / len(building_prices)
            if building_prices else 0
        ),
        "parking_avg_price": (
            sum(p.total_price for p in parking_prices) / len(parking_prices)
            if parking_prices else 0
        ),
    }