470 lines
17 KiB
Python
470 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.earth_news import EarthNewsItem
|
|
from app.services.earth_news import (
|
|
ParsedNewsItem,
|
|
apply_enrichment_patch_to_item,
|
|
build_anchor_location_patch,
|
|
_news_meta_patch,
|
|
)
|
|
from app.services.earth_news_classification import (
|
|
breaking_sort_rank,
|
|
normalize_breaking_level,
|
|
normalize_breaking_scope,
|
|
)
|
|
|
|
CRUISE_REGION_ORDER = (
|
|
"americas",
|
|
"europe",
|
|
"middle-east-africa",
|
|
"asia-pacific",
|
|
"global",
|
|
)
|
|
CRUISE_REGION_QUERY_MULTIPLIER = 12
|
|
CRUISE_REGION_QUERY_MIN_LIMIT = 240
|
|
CRUISE_REGION_QUERY_MAX_LIMIT = 1000
|
|
|
|
|
|
def _coerce_datetime(value: datetime | None) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=UTC)
|
|
return value.astimezone(UTC)
|
|
|
|
|
|
def _coerce_meta_datetime(value: Any) -> datetime | None:
|
|
if isinstance(value, datetime):
|
|
return _coerce_datetime(value)
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
try:
|
|
return _coerce_datetime(datetime.fromisoformat(value.replace("Z", "+00:00")))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _location_patch_from_record(record: EarthNewsItem) -> dict[str, Any]:
|
|
return {
|
|
"latitude": record.latitude,
|
|
"longitude": record.longitude,
|
|
"location_label": record.location_label,
|
|
"location_source": record.location_source,
|
|
"verified": record.verified,
|
|
"location_meta": dict(record.location_meta or {}),
|
|
}
|
|
|
|
|
|
def record_to_parsed_news_item(record: EarthNewsItem) -> ParsedNewsItem:
|
|
location_meta = dict(record.location_meta or {})
|
|
news_meta = location_meta.get("news_meta") if isinstance(location_meta.get("news_meta"), dict) else {}
|
|
item = ParsedNewsItem(
|
|
id=record.id,
|
|
title=record.title,
|
|
summary=record.summary or "",
|
|
url=record.url,
|
|
source=record.source or "",
|
|
feed_name=record.feed_name or "",
|
|
feed_region=record.region or "global",
|
|
homepage_url=record.homepage_url or "",
|
|
published_at=_coerce_datetime(record.published_at),
|
|
content_language=record.content_language or "en",
|
|
localizations=dict(record.localizations or {}),
|
|
enrichment_status=record.enrichment_status or "pending",
|
|
enrichment_error=record.enrichment_error,
|
|
enriched_at=_coerce_datetime(record.enriched_at),
|
|
source_tags=list(news_meta.get("source_tags") or []),
|
|
feed_id=str(news_meta.get("feed_id") or ""),
|
|
feed_type=str(news_meta.get("feed_type") or "rss"),
|
|
feed_default_category=str(news_meta.get("feed_default_category") or "other"),
|
|
category=str(news_meta.get("category") or "other"),
|
|
item_tags=list(news_meta.get("item_tags") or []),
|
|
tagging_source=str(news_meta.get("tagging_source") or "rules"),
|
|
tagging_confidence=float(news_meta.get("tagging_confidence") or 0),
|
|
importance_score=int(news_meta.get("importance_score") or 0),
|
|
importance_level=str(news_meta.get("importance_level") or "low"),
|
|
importance_reasons=list(news_meta.get("importance_reasons") or []),
|
|
market_impact=str(news_meta.get("market_impact") or "none"),
|
|
breaking_level=normalize_breaking_level(news_meta.get("breaking_level")).value,
|
|
breaking_scope=normalize_breaking_scope(news_meta.get("breaking_scope")).value,
|
|
breaking_reasons=list(news_meta.get("breaking_reasons") or []),
|
|
breaking_source=str(news_meta.get("breaking_source") or "rules"),
|
|
breaking_confidence=float(news_meta.get("breaking_confidence") or 0),
|
|
breaking_expires_at=_coerce_meta_datetime(news_meta.get("breaking_expires_at")),
|
|
)
|
|
return apply_enrichment_patch_to_item(item, _location_patch_from_record(record))
|
|
|
|
|
|
def _sort_parsed_news_items(items: list[ParsedNewsItem], *, active_region: str) -> list[ParsedNewsItem]:
|
|
return sorted(
|
|
items,
|
|
key=lambda item: (
|
|
-breaking_sort_rank(item),
|
|
False
|
|
if active_region == "global"
|
|
or (breaking_sort_rank(item) > 0 and normalize_breaking_scope(item.breaking_scope).value == "global")
|
|
else item.feed_region != active_region,
|
|
item.published_at is None,
|
|
-(item.published_at.timestamp() if item.published_at else 0),
|
|
item.feed_name,
|
|
),
|
|
)
|
|
|
|
|
|
def _diversify_parsed_news_items_by_region(
|
|
items: list[ParsedNewsItem],
|
|
*,
|
|
limit: int,
|
|
) -> list[ParsedNewsItem]:
|
|
if limit <= 0:
|
|
return []
|
|
sorted_items = _sort_parsed_news_items(items, active_region="global")
|
|
buckets: dict[str, list[ParsedNewsItem]] = {}
|
|
for item in sorted_items:
|
|
region = item.feed_region or "global"
|
|
buckets.setdefault(region, []).append(item)
|
|
|
|
ordered_regions = [
|
|
*[region for region in CRUISE_REGION_ORDER if buckets.get(region)],
|
|
*sorted(region for region in buckets if region not in CRUISE_REGION_ORDER),
|
|
]
|
|
diversified: list[ParsedNewsItem] = []
|
|
cursor = 0
|
|
while len(diversified) < limit:
|
|
added = False
|
|
for region in ordered_regions:
|
|
bucket = buckets.get(region) or []
|
|
if cursor >= len(bucket):
|
|
continue
|
|
diversified.append(bucket[cursor])
|
|
added = True
|
|
if len(diversified) >= limit:
|
|
break
|
|
if not added:
|
|
break
|
|
cursor += 1
|
|
return diversified
|
|
|
|
|
|
def _query_sort_key(active_region: str):
|
|
if active_region == "global":
|
|
return (
|
|
EarthNewsItem.published_at.is_(None),
|
|
EarthNewsItem.published_at.desc().nullslast(),
|
|
EarthNewsItem.feed_name.asc(),
|
|
)
|
|
return (
|
|
EarthNewsItem.region != active_region,
|
|
EarthNewsItem.published_at.is_(None),
|
|
EarthNewsItem.published_at.desc().nullslast(),
|
|
EarthNewsItem.feed_name.asc(),
|
|
)
|
|
|
|
|
|
def _category_filter_clause(categories: set[str] | None):
|
|
if not categories:
|
|
return None
|
|
return EarthNewsItem.location_meta.op("->")("news_meta").op("->>")("category").in_(sorted(categories))
|
|
|
|
|
|
def _source_filter_clause(source_ids: set[str] | None):
|
|
if not source_ids:
|
|
return None
|
|
return EarthNewsItem.location_meta.op("->")("news_meta").op("->>")("source_id").in_(sorted(source_ids))
|
|
|
|
|
|
async def list_earth_news_items(
|
|
db: AsyncSession,
|
|
*,
|
|
active_region: str,
|
|
limit: int,
|
|
categories: set[str] | None = None,
|
|
source_ids: set[str] | None = None,
|
|
) -> list[ParsedNewsItem]:
|
|
query_limit = limit if source_ids else min(max(limit * 20, limit), 500)
|
|
query = (
|
|
select(EarthNewsItem)
|
|
.order_by(*_query_sort_key(active_region))
|
|
.limit(query_limit)
|
|
)
|
|
if active_region != "global":
|
|
news_meta = EarthNewsItem.location_meta.op("->")("news_meta")
|
|
query = query.where(
|
|
or_(
|
|
EarthNewsItem.region.in_({"global", active_region}),
|
|
news_meta.op("->>")("breaking_scope") == "global",
|
|
)
|
|
)
|
|
category_clause = _category_filter_clause(categories)
|
|
if category_clause is not None:
|
|
query = query.where(category_clause)
|
|
source_clause = _source_filter_clause(source_ids)
|
|
if source_clause is not None:
|
|
query = query.where(source_clause)
|
|
result = await db.execute(query)
|
|
records = list(result.scalars().all())
|
|
items = _sort_parsed_news_items(
|
|
[record_to_parsed_news_item(record) for record in records],
|
|
active_region=active_region,
|
|
)
|
|
if active_region == "global" and not source_ids:
|
|
return _diversify_parsed_news_items_by_region(items, limit=limit)
|
|
return items[:limit]
|
|
|
|
|
|
async def list_earth_news_cruise_items(
|
|
db: AsyncSession,
|
|
*,
|
|
limit: int,
|
|
categories: set[str] | None = None,
|
|
source_ids: set[str] | None = None,
|
|
) -> list[ParsedNewsItem]:
|
|
query_limit = min(
|
|
max(limit * CRUISE_REGION_QUERY_MULTIPLIER, CRUISE_REGION_QUERY_MIN_LIMIT),
|
|
CRUISE_REGION_QUERY_MAX_LIMIT,
|
|
)
|
|
query = (
|
|
select(EarthNewsItem)
|
|
.order_by(
|
|
EarthNewsItem.published_at.is_(None),
|
|
EarthNewsItem.published_at.desc().nullslast(),
|
|
EarthNewsItem.last_seen_at.desc(),
|
|
EarthNewsItem.region.asc(),
|
|
EarthNewsItem.feed_name.asc(),
|
|
)
|
|
.limit(query_limit)
|
|
)
|
|
category_clause = _category_filter_clause(categories)
|
|
if category_clause is not None:
|
|
query = query.where(category_clause)
|
|
source_clause = _source_filter_clause(source_ids)
|
|
if source_clause is not None:
|
|
query = query.where(source_clause)
|
|
result = await db.execute(query)
|
|
return _diversify_parsed_news_items_by_region(
|
|
[record_to_parsed_news_item(record) for record in result.scalars().all()],
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
async def get_earth_news_freshness(
|
|
db: AsyncSession,
|
|
*,
|
|
active_region: str,
|
|
) -> tuple[int, datetime | None]:
|
|
query = select(
|
|
func.count(EarthNewsItem.id),
|
|
func.max(func.coalesce(EarthNewsItem.published_at, EarthNewsItem.last_seen_at)),
|
|
)
|
|
if active_region != "global":
|
|
query = query.where(EarthNewsItem.region.in_({"global", active_region}))
|
|
result = await db.execute(query)
|
|
count, newest = result.one()
|
|
item_count = int(count or 0)
|
|
if item_count == 0:
|
|
return 0, None
|
|
return item_count, _coerce_datetime(newest)
|
|
|
|
|
|
async def get_earth_news_feed_coverage(
|
|
db: AsyncSession,
|
|
*,
|
|
active_region: str,
|
|
recent_after: datetime | None = None,
|
|
) -> set[tuple[str, str]]:
|
|
query = select(
|
|
EarthNewsItem.id,
|
|
EarthNewsItem.location_meta.op("->")("news_meta").op("->>")("source_id"),
|
|
EarthNewsItem.location_meta.op("->")("news_meta").op("->>")("feed_id"),
|
|
)
|
|
if active_region != "global":
|
|
query = query.where(EarthNewsItem.region.in_({"global", active_region}))
|
|
if recent_after is not None:
|
|
query = query.where(func.coalesce(EarthNewsItem.published_at, EarthNewsItem.last_seen_at) >= recent_after)
|
|
result = await db.execute(query)
|
|
coverage: set[tuple[str, str]] = set()
|
|
for item_id, source_id, feed_id in result.all():
|
|
normalized_source_id = str(source_id or "").strip()
|
|
normalized_feed_id = str(feed_id or "").strip()
|
|
if not normalized_source_id and isinstance(item_id, str) and ":" in item_id:
|
|
normalized_source_id = item_id.split(":", 1)[0]
|
|
if normalized_source_id and normalized_feed_id:
|
|
coverage.add((normalized_source_id, normalized_feed_id))
|
|
return coverage
|
|
|
|
|
|
async def upsert_earth_news_items(db: AsyncSession, items: list[ParsedNewsItem]) -> int:
|
|
if not items:
|
|
return 0
|
|
now = datetime.now(UTC)
|
|
existing_result = await db.execute(
|
|
select(EarthNewsItem).where(EarthNewsItem.id.in_([item.id for item in items]))
|
|
)
|
|
existing = {record.id: record for record in existing_result.scalars().all()}
|
|
changed = 0
|
|
for item in items:
|
|
record = existing.get(item.id)
|
|
if record is None:
|
|
patch = build_anchor_location_patch(item)
|
|
record = EarthNewsItem(
|
|
id=item.id,
|
|
title=item.title,
|
|
summary=item.summary,
|
|
content_language=item.content_language,
|
|
localizations=dict(item.localizations or {}),
|
|
url=item.url,
|
|
source=item.source,
|
|
feed_name=item.feed_name,
|
|
region=item.feed_region,
|
|
homepage_url=item.homepage_url,
|
|
published_at=item.published_at,
|
|
latitude=patch["latitude"],
|
|
longitude=patch["longitude"],
|
|
location_label=patch["location_label"],
|
|
location_source=patch["location_source"],
|
|
verified=patch["verified"],
|
|
location_meta=patch["location_meta"],
|
|
first_seen_at=now,
|
|
last_seen_at=now,
|
|
enrichment_status=item.enrichment_status,
|
|
enrichment_error=item.enrichment_error,
|
|
enriched_at=item.enriched_at,
|
|
)
|
|
db.add(record)
|
|
changed += 1
|
|
continue
|
|
|
|
record.title = item.title
|
|
record.summary = item.summary
|
|
record.url = item.url
|
|
record.source = item.source
|
|
record.feed_name = item.feed_name
|
|
record.region = item.feed_region
|
|
record.homepage_url = item.homepage_url
|
|
record.published_at = item.published_at
|
|
record.last_seen_at = now
|
|
location_meta = dict(record.location_meta or {})
|
|
location_meta["news_meta"] = _news_meta_patch(item)
|
|
record.location_meta = location_meta
|
|
if item.localizations:
|
|
merged_localizations = {
|
|
**dict(record.localizations or {}),
|
|
**dict(item.localizations or {}),
|
|
}
|
|
record.content_language = item.content_language
|
|
record.localizations = merged_localizations
|
|
if item.enrichment_status != "pending" or item.enrichment_error or item.enriched_at:
|
|
record.enrichment_status = item.enrichment_status
|
|
record.enrichment_error = item.enrichment_error
|
|
record.enriched_at = item.enriched_at
|
|
changed += 1
|
|
await db.flush()
|
|
return changed
|
|
|
|
|
|
async def update_earth_news_item_location(
|
|
db: AsyncSession,
|
|
*,
|
|
item_id: str,
|
|
patch: dict[str, Any],
|
|
) -> bool:
|
|
record = await db.get(EarthNewsItem, item_id)
|
|
if record is None:
|
|
return False
|
|
record.latitude = float(patch["latitude"])
|
|
record.longitude = float(patch["longitude"])
|
|
record.location_label = str(patch["location_label"])
|
|
record.location_source = str(patch["location_source"])
|
|
record.verified = bool(patch["verified"])
|
|
record.location_meta = dict(patch.get("location_meta") or {})
|
|
record.resolved_at = datetime.now(UTC) if record.verified else None
|
|
await db.flush()
|
|
return True
|
|
|
|
|
|
async def update_earth_news_item_enrichment(
|
|
db: AsyncSession,
|
|
*,
|
|
item_id: str,
|
|
patch: dict[str, Any],
|
|
) -> bool:
|
|
record = await db.get(EarthNewsItem, item_id)
|
|
if record is None:
|
|
return False
|
|
if "latitude" in patch:
|
|
patch_meta = dict(patch.get("location_meta") or {})
|
|
if record.location_source == "manual_location":
|
|
current_meta = dict(record.location_meta or {})
|
|
patch_news_meta = patch_meta.get("news_meta")
|
|
if isinstance(patch_news_meta, dict):
|
|
current_meta["news_meta"] = patch_news_meta
|
|
current_meta["manual_enrichment"] = {
|
|
"resolution_stage": patch_meta.get("resolution_stage"),
|
|
"ai_attempted": patch_meta.get("ai_attempted"),
|
|
"ai_status": patch_meta.get("ai_status"),
|
|
"ai_error": patch_meta.get("ai_error"),
|
|
"debug_note": patch_meta.get("debug_note"),
|
|
}
|
|
record.location_meta = current_meta
|
|
else:
|
|
record.latitude = float(patch["latitude"])
|
|
record.longitude = float(patch["longitude"])
|
|
record.location_label = str(patch["location_label"])
|
|
record.location_source = str(patch["location_source"])
|
|
record.verified = bool(patch["verified"])
|
|
record.location_meta = patch_meta
|
|
record.resolved_at = datetime.now(UTC) if record.verified else None
|
|
if "content_language" in patch:
|
|
record.content_language = str(patch.get("content_language") or "en")
|
|
if "localizations" in patch:
|
|
record.localizations = dict(patch.get("localizations") or {})
|
|
if "enrichment_status" in patch:
|
|
record.enrichment_status = str(patch.get("enrichment_status") or "pending")
|
|
if "enrichment_error" in patch:
|
|
record.enrichment_error = patch.get("enrichment_error")
|
|
if patch.get("enriched_at"):
|
|
try:
|
|
parsed_enriched_at = datetime.fromisoformat(
|
|
str(patch["enriched_at"]).replace("Z", "+00:00")
|
|
)
|
|
except ValueError:
|
|
parsed_enriched_at = datetime.now(UTC)
|
|
record.enriched_at = _coerce_datetime(parsed_enriched_at)
|
|
elif patch.get("localizations"):
|
|
record.enriched_at = datetime.now(UTC)
|
|
await db.flush()
|
|
return True
|
|
|
|
|
|
async def list_unverified_earth_news_items(
|
|
db: AsyncSession,
|
|
*,
|
|
active_region: str,
|
|
limit: int,
|
|
) -> list[ParsedNewsItem]:
|
|
regions = {"global", active_region}
|
|
result = await db.execute(
|
|
select(EarthNewsItem)
|
|
.where(EarthNewsItem.region.in_(regions))
|
|
.where(EarthNewsItem.verified.is_(False))
|
|
.order_by(*_query_sort_key(active_region))
|
|
.limit(limit)
|
|
)
|
|
return [record_to_parsed_news_item(record) for record in result.scalars().all()]
|
|
|
|
|
|
async def list_all_earth_news_records(db: AsyncSession) -> list[EarthNewsItem]:
|
|
result = await db.execute(
|
|
select(EarthNewsItem).order_by(
|
|
EarthNewsItem.published_at.desc().nullslast(),
|
|
EarthNewsItem.last_seen_at.desc(),
|
|
)
|
|
)
|
|
return list(result.scalars().all())
|