HardSeat-Hero项目搭建流程记录

class TrainDataProvider:
    name: str

    async def search_segments(
        self,
        from_station: str,
        to_station: str,
        query: RouteQuery,
    ) -> list[TrainSegment]:
        ...

    def candidate_transfer_stations(self, query: RouteQuery) -> list[str]:
        ...

# 称作贫血模型，纯粹和表结构一模一样，为了装数据而产生
class UserTable:
    user_id: int
    score: int

class PostTable:
    post_id: int
    user_id: int
    content: str


# 所有的业务逻辑都挤在Service层的一个函数中
def publish_post_service(user_id: int, post_id: int):
    # 查表
    user_data = db.query(UserTable).filter_by(user_id=user_id).first()

    # 在这里做校验
    if len(content) <= 100:
        raise HTTPException(status_code=400, detail="太短了")

    # 强行修改字段
    user_data.score += 10
    new_post = PostTable(user_id=user_id, content=content)

    db.add(new_post)
    db.commit()

from dataclasses import dataclass
from typing import Optional

# 值对象
@dataclass(frozen=True)     # 值对象不可变
class PostContent:
    value: str

    def __post_init__(self):
        if len(self.value) <= 100:
            raise ValueError(f"文章内容太短了，当前{len(self.value)}个字")


# 实体与聚合根
class User:
    def __init__(self, user_id: int, score: int, level: int = 1):
        self.id = user_id
        self._score = score
        self._level = level

    @property
    def score(self) -> int:
        return self._score

    # 业务行为
    def calculate_publish_reward(self) -> int:
        if self._level >= 5:
            return 20
        return 10

    def add_score(self, amount: int):
        if amount < 0:
            raise ValueError("积分不能为负数")
        self._score += amount


class Post:
    def __init__(self, post_id: Optional[int], author_id: int, content: PostContent):
        self.id = post_id
        self.author_id = author_id
        self.content = content

from abc import ABC, abstractmethod


class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        pass


class PostRepository(ABC):
    @abstractmethod
    def save(self, post: Post) -> None:
        pass

class PublishPostApplicationService:
    def __init__(self, user_repo: UserRespository, post_repo: PostRepository):
        self.user_repo = user_repo
        self.post_repo = post_repo

    def execute(self, user_id: int, content_text: str) -> None:
        user = self.user_repo.find_by_id(user_id)
        if not user:
            raise Exception("用户不存在")

        # 依赖值对象自校验
        content = PostContent(value=content_text)
        post = Post(post_id=None, author_id=user.id, content=content)

        # 领域对象之间的业务交互
        reward_score = user.calculate_publish_reward()
        user.add_score(reward_score)

        # 持久化保存
        self.post_repo.save(post)
        self.user_repo.save(user)

class SeatPrice(BaseModel):
    seat_type: str
    price: Decimal - Field(ge=0)
    remaining: str = "unknown"

class TrainSegment(BaseModel):
    train_no: str
    from_station: str
    to_station: str
    depart_at: datetime
    arrive_at: datetime
    duration_minutes: int = Field(gt=0)     # 字段限定
    prices: list[SeatPrice]
    source: str = "mock"
    updated_at: datetime

    @model_validator(mode="after")      # 值对象自校验
    def validate_time_order(self) -> "TrainSegment":    # 前向引用，防止此时类未定义完报错
        if self.arrive_at <= self.depart_at:
            raise ValueError("arrive_at must be later than depart_at")
        return self

    @property   # 用于读取最低价格，该装饰器可使方法像变量一样调用，且只读
    def lowest_price(self) -> Decimal | None:
        if not self.prices:
            return None
        return min(price.price for price in self.prices)

class RouteQuery(BaseModel):
    from_station: str = Field(min_length=1, max_length=64)
    to_station: str = Field(min_length=1, max_length=64)
    date: date
    max_transfers: int = Field(default=1, ge=0, le=2)               # 最多中转次数
    min_transfer_minutes: int = Field(default=30, ge=0, le=360)     # 需求的最少中转时间
    max_total_duration_minutes: int = Field(default=48 * 60, ge=60, le=72 * 60) # 全程最多花费时间

    @field_validator("from_station", "to_station", mode="before")
    @classmethod    # 类方法
    def strip_station_name(cls, value: Any) -> str:
        if isinstance(value, str):
            return value.strip()
        return value

    @model_validator(mode="after")
    def validate_distinct_stations(self) -> "RouteQuery":
        if self.from_station == self.to_station:
            raise ValueError("from_station and to_station must be different")
        return self

class TransferPlan(BaseModel):
    total_price: Decimal
    total_duration_minutes: int
    transfer_minutes: int
    transfer_stations: list[str]
    segments: list[TrainSegment]

class RouteSearchResponse(BaseModel):
    query_id: str
    source: str
    updated_at: datetime
    plans: list[TransferPlan]

class StationSearchResponse(BaseModel):
    stations: list[str]

class StationMetadata(BaseModel):
    name: str
    telecode: str
    latitude: float
    longitude: float
    centrality_score: float = Field(default=0, ge=0)

@router.get("/stations/search", response_model=StationSearchResponse)
async def search_stations(q: str = "") -> StationSearchResponse:
    search = getattr(provider, "search_stations", None)     # 动态获取方法
    if search is None:      # 方法不存在
        return StationSearchResponse(stations=[])
    
    try:        # 防止search(q)抛异常直接冒泡到500 Interval Server Error
        stations = search(q)
        if hasattr(stations, "__await__"):
            stations = await stations
        return StationSearchResponse(stations=stations)
    except TrainDataProviderError as exc:
        raise HTTPException(status_code=provider_error_status_code(exc), detail=str(exc)) from exc

class MockTrainDataProvider(TrainDataProvider):
    name = "mock"

    def __init__(self) -> None:
        self._stations = ["北京", "济南西", "南京南", "上海", "天津南", "杭州东"]
    ...
    def search_stations(self, keyword: str) -> list[str]:
        if not keyword:
            return self._stations
        return [station for station in self._stations if keyword in station]

class Railway12306PublicPriceProvider(TrainDataProvider):
    name = "12306-public-price"
    ...
    async def search_stations(self, keyword: str) -> list[str]:
        return await self.station_repository.search_stations(keyword)

...

class StationCodeRepository:
    ...
    async def search_stations(self, keyword: str) -> list[str]:
        stations = sorted((await self.get_station_codes()).keys())
        if not keyword:
            return stations[:20]
        return [station for station in stations if keyword in station][:20]

class LocalStaticPriceProvider(TrainDataProvider):
    name = "static-price"
    ...
    # 没有search_stations的实现

class StaticPriceFallbackProvider(TrainDataProvider):
    name = "static-price"
    ...
    # 同样没有search_stations的实现

provider + date + from_station + to_station -> list[TrainSegment]

class TtlCache(Generic[T]):     # 泛型类
    def __init__(self, ttl_seconds: int) -> None:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be greater than 0")
        self.ttl = timedelta(seconds=ttl_seconds)
        self._items: dict[str, tuple[datetime, T]] = {}

    def get(self, key: str) -> T | None:
        item = self._items.get(key)
        if item is None:
            return None
        
        expires_at, value = item
        if datetime.now(timezone.utc) >= expires_at:    # 过期删除，这里没有用到主动随机删除
            self._items.pop(key, None)
            return None

        return value

    def set(self, key: str, value: T) -> None:
        self._items[key] = (datetime.now(timezone.utc) + self.ttl, value)

    def clear(self) -> None:
        self._items.clear()

某日 + 出发站 + 到达站

GET https://kyfw.12306.cn/otn/leftTicketPrice/queryAllPublicPrice

07950 -> 795.0
01145 -> 114.5

class Railway12306PublicPriceProvider(TrainDataProvider):
    name = "12306-public-price"

    def __init__(
        self,
        client: PublicPriceQueryClient | None = None,
        transfer_generator: CandidataTransferStationGenerator | None = None,
    ) -> None:
        station_repository = StationCodeRepository()
        self.client = client or PublicPriceClient(station_repository)
        self.station_repository = station_repository
        self.transfer_generator = transfer_generator or CandidateTransferStationGenerator()

    async def search_segments(
        self,
        from_station: str,
        to_station: str,
        query: RouteQuery,
    ) -> list[TrainSegment]:
        updated_at = datetime.now(timezone.utc)
        rows = await self.client.query(from_station, to_station, query)
        segments = [build_segment_from_public_price(row, query, updated_at) for row in rows]
        return [segment for segment in segments if segment is not None]

    def candidate_transfer_stations(self, query: RouteQuery) -> list[str]:
        return self.transfer_generator.generate(query)

    async def search_stations(self, keyword: str) -> list[str]:
        return await self.station_repository.search_stations(keyword)

def describe_httpx_error(exc: httpx.HTTPError) -> str:
    if isinstance(exc, httpx.HTTPStatusError):
        response = exc.response
        return f"{type(exc).__name__}: HTTP {response.status_code} {response.reason_phrase}"
    if isinstance(exc, httpx.TimeoutException):
        return f"{type(exc).__name__}: 请求超时"
    if isinstance(exc, httpx.RequestError):
        request = exc.request
        return f"{type(exc).__name__}: {request.method} {request.url} 请求失败：{exc}"
    return f"{type(exc).__name__}: {exc}"


class StationCodeRepository:
    def __init__(self, timeout_seconds: float = 10.0, cache_ttl_seconds: int = 86400) -> None:
        self.timeout_seconds = timeout_seconds
        self.cache: TtlCache[dict[str, str]] = TtlCache(cache_ttl_seconds)

    async def get_station_codes(self) -> dict[str, str]:
        cached = self.cache.get("station-codes")
        if cached is not None:
            return cached

        last_error: Exception | None = None
        async with httpx.AsyncClient(timeout=self.timeout_seconds, headers=request_headers()) as client:
            for url in STATION_URLS:
                try:
                    response = await client.get(url)
                    response.raise_for_status()
                    station_codes = parse_station_codes(response.text)
                    self.cache.get("station-codes", station_codes)
                    return station_codes
                except (httpx.HTTPError, Railway12306Error) as exc:
                    last_error = exc
        if isinstance(last_error, httpx.HTTPError):
            detail = describe_httpx_error{last_error}"
        else:
            detail = f"{type{last_error}.__name__}: {last_error}"
        raise Railway12306Error(f"无法获取车站电报码：{detail}")

    async def search_stations(self, keyword: str) -> list[str]:
        stations = sorted((await self.get_station_codes()).keys())
        if not keyword:
            return stations[:20]
        return [station for station in stations if keyword in station][:20]

class PublicPriceQueryClient(Protocol):     # 结构化接口，使得下面的PublicPriceClient符合PublicPriceQueryClient协议，测试时可以传入同样实现query方法的假客户端，从而不必真的访问12306接口
    async def query(self, from_station: str, to_station: str, query: RouteQuery) -> list[dict[str, Any]]:
        ...


class PublicPriceClient:
    def __init__(self, station_code: StationCodeRepository, timeout_seconds: float = 10.0) -> None:
        self.station_codes = station_codes
        self.timeout_seconds = timeout_seconds

    async def query(self, from_station: str, to_station: str, query: RouteQuery) -> list[dict[str, Any]]:
        station_codes = await self.station_codes.get_station_codes()
        try:
            from_code = station_codes[from_station]
            to_code = station_codes[to_station]
        except KeyError as exc:
            raise Railway12306Error(f"站名无法映射电报码：{exc}") from exc

        params = {
            "leftTicketDTO.train_date": query.date.isoformat(),
            "leftTicketDTO.from_station": from_code,
            "leftTicketDTO.to_station": to_code,
            "purpose_codes": "ADULT",
        }
        async with httpx.AsyncClient(timeout=self.timeout_seconds, headers=request_headers()) as client:
            try:
                await client.get(PUBLIC_QUERY_PAGE_URL)
                response = await client.get(PUBLIC_PRICE_URL, params=params)
                response.raise_for_status()     # 检测并抛出HTTP异常
            except httpx.HTTPError as exc:
                raise Railway12306Error(f"票价接口请求失败：{describe_httpx_error(exc)}") from exc

        try:
            payload = response.json()
        except ValueError as exc:
            raise Railway12306Error("票价接口返回非 JSON") from exc
        if not payload.get("status"):
            raise Railway12306Error(f"票价接口返回失败：{payload}")
        rows = payload.get("data", [])
        if not isinstance(rows, list):
            raise Railway12306Error("票价接口 data 不是列表")
        return rows

leftTicketDTO.train_date=YYYY-MM-DD
leftTicketDTO.from_station=出发站电报码
leftTicketDTO.to_station=到达站电报码
purpose_codes=ADULT

{
  "validateMessagesShowId": "_validatorMessage",
  "status": true,
  "httpstatus": 200,
  "data": [
    {
      "queryLeftNewDTO": {
        "train_no": "240000G54700",
        "station_train_code": "G547",
        "train_class_name": "高速",
        "from_station_name": "北京南",
        "to_station_name": "上海虹桥",
        "from_station_telecode": "VNP",
        "to_station_telecode": "AOH",
        "start_time": "06:18",
        "arrive_time": "12:11",
        "lishi": "05:53",
        "day_difference": "0",
        "swz_price": "27820",
        "zy_price": "12720",
        "ze_price": "07950"
      }
    }
  ],
  "messages": [],
  "validateMessages": {}
}

PRICE_FIELDS = {
    "swz_price": "商务座",
    "zy_price": "一等座",
    "ze_price": "二等座",
    "yz_price": "硬座",
    "yw_price": "硬卧",
    "rw_price": "软卧",
    "gr_price": "高级软卧",
}


def parse_duration_minutes(value: str) -> int:
    parts = value.split(":")
    if len(parts) != 2:
        raise Railway12306Error(f"历时格式不符合预期：{value}")
    hours, minutes = (int(part) for part in parts)
    return hours * 60 + minutes


def parse_day_difference(value: Any) -> int:
    if value in (None, ""):
        return 0
    return int(value)


def build_segment_from_public_price(row: dict[str, Any], query: RouteQuery, updated_at: datetime) -> TrainSegment | None:
    dto = row.get("queryLeftNewDTO")
    if not isinstance(dto, dict):
        raise Railway12306Error("缺少 queryLeftNewDTO")

    prices = [
        SeatPrice(seat_type=seat_type, price=price, remaining="unknown")
        for field, seat_type in PRICE_FIELDS.items()
        if (price := parse_price(dto.get(field))) is not None   # 海象运算符，用于一边解析field赋值给price一边对price进行判断
    ]
    if not prices:
        return None

    # 解析逻辑
    depart_time = time.fromisoformat(str(dto.get("start_time", "")))
    arrive_time = time.fromisoformat(str(dto.get("arrive_time", "")))
    day_difference = parse_day_difference(dto.get("day_difference"))
    depart_at = datetime.combine(query.date, depart_time, tzinfo=timezone.utc)
    arrive_at = datetime.combine(query.date + timedelta(days=day_difference), arrive_time, tzinfo=timezone.utc)
    duration_minutes = parse_duration_minutes(str(dto.get("lishi", "")))

    # 转为TrainSegment返回
    return TrainSegment(
        train_no=str(dto.get("station_train_code") or dto.get("train_no") or ""),
        from_station=str(dto.get("from_station_name") or query.from_station),
        to_station=str(dto.get("to_station_name") or query.to_station),
        depart_at=depart_at,
        arrive_at=arrive_at,
        duration_minutes=duration_minutes,
        prices=prices,
        source="12306-public-price",
        updated_at=updated_at,
    )

def create_train_data_provider() -> TrainDataProvider:
    provider_name = os.getenv("TRAIN_DATA_PROVIDER", "mock").strip().lower()
    if provider_name == "mock":
        return MockTrainDataProvider()
    if provider_name in {"12306", "12306-public-price", "railway_12306_public_price"}:
        return Railway12306PublicPriceProvider()
    if provider_name in {"static", "static-price", "local-static-price"}:
        return _create_static_price_provider()
    raise ValueError(f"Unsupported TRAIN_DATA_PROVIDER: {provider_name}")

class CandidateTranferStationGenerator:
    def __init__(
        self,
        repository: StationMetadataRepository | None = None,
        config: CandidataTransferConfig | None = None
    ) -> None:
        self.repository = repository or StationMetadataRepository()
        self.config = config or CandidateTransferConfig()

    def generate(self, query: RouteQuery) -> list[str]:
        # 换乘枢纽和铁路通道先验
        origin = self.repository.get(query.from_station)
        destination = self.repository.get(query.to_station)
        if origin is None or destination is None:
            return self._fallback_low_price_hubs(query)

        # 地理走廊过滤
        ab_distance = haversin_km(origin, destination)
        if ab_distance == 0:
            return []

        # 定义走廊距离
        corridor_km = max(
            self.config.min_corrider_km,
            min(self.config.max_corridor_km, self.config.corridor_ratio * ab_distance),
        )
        scored: list[tuple[tuple[float, float, float, str], str]] = []
        for station in self.repository.all():
            if station.name in [origin.name, destination.name]:
                continue

            # 投影方向过滤
            projection, perpendicular = project_station(origin, destination, station)
            projection_margin_km = ab_distance * 0.25
            if projection < -projection_margin_km or projection > ab_distance + projection_margin_km:
                continue
            if projection < self.config.endpoint_exclusion_km or projection > ab_distance - self.config.endpoint_exclusion_km:
                continue

            # 走廊宽度过滤
            cheap_transfer_score = self._cheap_transfer_score(station.name)
            if cheap_transfer_score > 0:
                max_perpendicular = max(corridor_km, min(500, ab_distance * 0.75))
                if perpendicular > max_perpendicular:
                    continue
            elif perpendicular > corridor_km:
                continue
            score = (-cheap_transfer_score, -station.centrality_score, perpendicular, station.name)
            scored.append((score, station.name))
        
        scored.sort(key=lambda item: item[0])
        return [name for _, name in scored[: self.config.max_candidates]]

    def _fallback_low_price_hubs(self, query: RouteQuery) -> list[str]:
        blocked = {query.from_station, query.to_station}
        hubs = [
            station.name
            for station in self.repository.all()
            if station.name in LOW_PRICE_TRANSFER_HUBS and station.name not in blocked
        ]
        return hubs[: self.config.max_candidates]

    def _cheap_transfer_score(self, station_name: str) -> float:
        if station_name in CHEAP_TRANSFER_SCORES:
            return CHEAP_TRANSFER_SCORES[station_name]
        if station_name in LOW_PRICE_TRANSFER_HUBS:
            return 50
        return 0

for first in first_legs:
    for second in second_legs:
        ...

class RouteQuery(BaseModel):
    from_station: str = Field(min_length=1, max_length=64)
    to_station: str = Field(min_length=1, max_length=64)
    date: date
    max_transfers: int = Field(default=1, ge=0, le=2)
    min_transfer_minutes: int = Field(default=30, ge=0, le=360)
    max_total_duration_minutes: int | None = Field(default=None, ge=60)
    candidate_limit: int = Field(default=50, ge=1, le=200)
    candidate_strategy: Literal["balanced", "direct_corridor", "wide_detour", "hub_first", "exhaustive_budgeted"] = "balanced"
    max_detour_ratio: float | None = Field(default=None, ge=0, le=3)
    max_detour_km: int | None = Field(default=None, ge=0, le=5000)
    corridor_width_km: int | None = Field(default=None, ge=1, le=2000)

    @field_validator("from_station", "to_station", mode="before")
    @classmethod
    def strip_station_name(cls, value: Any) -> str:
        if isinstance(value, str):
            return value.strip()
        return value

    @model_validator(mode="after")
    def validate_distinct_stations(self) -> "RouteQuery":
        if self.from_station == self.to_station:
            raise ValueError("from_station and to_station must be different")
        return self

class TransferPlan(BaseModel):
    total_price: Decimal
    total_duration_minutes: int
    transfer_minutes: int
    transfer_stations: list[str]
    segments: list[TrainSegment]

async def _iter_plans(self, query: RouteQuery, context: SearchContext) -> AsyncIterator[TransferPlan]:
    if query.max_transfers >= 2:
        for plan in await self.cheapest_path_searcher.search(query, context):
            yield plan
        return

    best_price = BestPriceTracker()
    direct_segments = await self._search_segments(query.from_station, query.to_station, query, context)
    ...

    if query.max_transfers >= 1:
        async for plan in self._iter_one_transfer(query, best_price, context):
            ...

max_transfers == 0  -> 只查直达
max_transfers == 1  -> 查直达 + 一次中转
max_transfers >= 2  -> 走多中转路径搜索 CheapestPathSearcher

direct_segments = await self._search_segments(query.from_station, query.to_station, query, context)
for segment in direct_segments:
    if segment.lowest_price is not None:
        plan = self._build_plan([segment])
        if self._plan_within_duration_limit(plan, query):
            await best_price.update(plan.total_price)
            yield plan

async def _iter_one_transfer(
    self,
    query: RouteQuery,
    best_price: BestPriceTracker,
    context: SearchContext,
) -> AsyncIterator[TransferPlan]:
    tasks = [
        asyncio.create_task(self._one_transfer_plans_for_station(query, transfer_station, best_price, context))
        for transfer_station in self._ranked_transfer_candidates(query)
    ]
    for task in asyncio.as_completed(tasks):
        ...

next_price + self._heuristic_lower_bound(...)

def _heuristic_lower_bound(self, from_station: str, to_station: str) -> Decimal:
    return Decimal("0")

@dataclass(frozen=True)
class PathSearchState:
    current_station: str
    segments: tuple[TrainSegment, ...]
    accumulated_price: Decimal
    arrive_at: datetime | None
    visited_stations: frozenset[str]

    @property
    def transfer_count(self) -> int:
        return max(0, len(self.segments) - 1)

queue: list[tuple[Decimal, int, PathSearchState]] = []
counter = 0
initial_state = PathSearchState(
    current_station=query.from_station,
    segments=(),
    accumulated_price=Decimal("0"),
    arrive_at=None,
    visited_stations=frozenset({query.from_station}),
)
heapq.heappush(queue, (Decimal("0"), counter, initial_state))

while queue:
    priority, _, state = heapq.heappop(queue)
    if best_price is not None and priority > best_price:
        break

def _next_stations(self, state: PathSearchState, query: RouteQuery, candidates: list[str]) -> list[str]:
    stations = [query.to_station]
    if state.transfer_count < query.max_transfers:
        stations.extend(station for station in candidates if station not in state.visited_stations)
    return stations

key = (state.current_station, state.transfer_count)
entries = self._entries.setdefault(key, [])
for entry in entries:
    if entry.price <= state.accumulated_price and entry.arrive_at <= state.arrive_at:
        return False

def _plan_sort_key(self, plan: TransferPlan) -> tuple[Decimal, int, int, int]:
    return (
        plan.total_price,
        self._transfer_wait_comfort_penalty(plan),
        len(plan.transfer_stations),
        plan.total_duration_minutes,
    )

if wait_minutes < COMFORTABLE_MIN_TRANSFER_MINUTES:
    penalty += (COMFORTABLE_MIN_TRANSFER_MINUTES - wait_minutes) * 10
elif wait_minutes > COMFORTABLE_MAX_TRANSFER_MINUTES:
    penalty += wait_minutes - COMFORTABLE_MAX_TRANSFER_MINUTES

tasks = [
    asyncio.create_task(self._one_transfer_plans_for_station(query, transfer_station, best_price, context))
    for transfer_staion in self._ranked_transfer_candidates(query)
]

for task in asyncio.as_completed(tasks):
    try:
        plans = await task
    except TrainDataProviderError:
        continue
    for plan in plans:
        yield plan

self._remote_query_semphore = asyncio.Seamphore(max_concurrent_remote_queries)

async with self._remote_query_semphore:
    segments = await self.provider.search_segments(from_station, to_station, query)

async with context.remote_query_lock:
    if context.remote_query_count >= self.max_remote_queries:
        return []

    should_wait = self.remote_query_interval_seconds > 0 and context.remote_query_count > 0
    context.remote_query_count += 1
    context.diagnostics.remote_query_count = context.remote_query_count

should_wait = self.remote_query_interval_seconds > 0 and context.remote_query_count > 0
context.remote_query_count += 1

if should_wait:
    await asyncio.sleep(self.remote_query_interval_seconds)

class BesttPriceTracker:
    def __init__(self) -> None:
        self._value: Decimal | None = None
        self._lock = asyncio.Lock()

    async def get(self) -> Decimal | None:
        async with self._lock:
            return self._value
        
    async def update(self, price: Decimal) -> None:
        async with self._lock:
            if self._value is None or self._value > price:
                self._value = price