Files
AlphaPulse/backend/src/alphapulse/news_risk/contracts.py
T
eezd 531ef724f3
Development CI / check-and-build (push) Successful in 3m17s
GitHub Git mirror / mirror (push) Successful in 6s
Development CI / deploy-development (push) Successful in 4s
feat(news): tune package research and add exclusive historical rechecks
2026-11-02 19:14:12 +08:00

59 lines
2.6 KiB
Python

"""Structured correction and time contracts shared by news stages."""
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, Field, model_validator
class PackageInput(BaseModel):
id: int = Field(gt=0)
revision: int = Field(ge=0)
class Assignment(BaseModel):
analysis_id: int = Field(gt=0)
evidence: str = Field(min_length=1, max_length=500)
class Replacement(BaseModel):
keep_id: int | None
title: str = Field(min_length=1, max_length=300)
fact: str = Field(min_length=1)
category: str = Field(min_length=1, max_length=64)
tags: list[str] = Field(min_length=1, max_length=20)
content_kind: Literal['report', 'statement', 'data', 'opinion', 'explanation', 'other']
assignments: list[Assignment] = Field(min_length=1)
class Correction(BaseModel):
packages: list[PackageInput] = Field(min_length=1)
replacements: list[Replacement] = Field(min_length=1)
reason: str = Field(min_length=1)
class PackageTimePoint(BaseModel):
occurred_at: datetime | None = Field(description="发生时间;未知必须为null且time_precision=unknown,已知必须带时区,不用报道时间代替")
time_precision: Literal['minute', 'day', 'month', 'unknown'] = Field(description='occurred_at为null时必须unknown;非null时按原文精度选择minute/day/month')
planned_at: datetime | None = Field(default=None, description='计划发生时间,不证明已发生')
planned_precision: Literal['minute', 'day', 'month', 'unknown'] = 'unknown'
period_text: str = Field(default='', description='数据统计期间原文')
time_text: str
reported_at: datetime | None
description: str = Field(min_length=1)
verification: Literal['supported', 'partial', 'conflicting', 'insufficient', 'refuted']
evidence_refs: list[dict] = Field(min_length=1)
@model_validator(mode='after')
def time_consistency(self):
if (self.planned_at is None) != (self.planned_precision == 'unknown'):
raise ValueError('计划时间与精度不一致')
if self.planned_at and self.planned_at.tzinfo is None:
raise ValueError('计划时间必须有时区')
if (self.occurred_at is None) != (self.time_precision == 'unknown'):
raise ValueError('发生时间与精度不一致')
if self.occurred_at and self.occurred_at.tzinfo is None:
raise ValueError('发生时间必须有时区')
if self.reported_at and self.reported_at.tzinfo is None:
raise ValueError('报道时间必须有时区')
return self