gitrusprusandCursor
699e9be503
Generate static HeuristicProfile in CA admin, preview and run without LLM on each post via shared interpreter in CP batch and listener. Co-authored-by: Cursor <cursoragent@cursor.com>
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""Per-source_type source_config schemas (CA admin + CP adapters)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
|
|
class TelegramSourceConfig(BaseModel):
|
|
channel: str = Field(min_length=1)
|
|
limit: int = Field(default=100, ge=1, le=1000)
|
|
# heuristic = legacy telegram_events; llm = DeepSeek per post; profile = static rules
|
|
extract_mode: Literal["heuristic", "llm", "profile"] = "heuristic"
|
|
extract_schema: dict[str, str] | None = None
|
|
instruction: str | None = None
|
|
heuristic_profile: dict | None = None
|
|
sample_post: str | None = None # audit / re-generate; not required at runtime
|
|
|
|
@field_validator("channel")
|
|
@classmethod
|
|
def strip_channel(cls, value: str) -> str:
|
|
return value.strip()
|
|
|
|
@model_validator(mode="after")
|
|
def profile_requires_rules(self) -> "TelegramSourceConfig":
|
|
if self.extract_mode != "profile":
|
|
return self
|
|
if not self.heuristic_profile:
|
|
raise ValueError("heuristic_profile required when extract_mode=profile")
|
|
from contracts.heuristic_profile import HeuristicProfile
|
|
|
|
HeuristicProfile.model_validate(self.heuristic_profile)
|
|
return self
|
|
|
|
|
|
class Crawl4AISourceConfig(BaseModel):
|
|
urls: list[str] = Field(min_length=1)
|
|
extract_schema: dict[str, str] = Field(
|
|
default_factory=lambda: {
|
|
"title": "string",
|
|
"locality": "string",
|
|
"event_date": "string",
|
|
"description": "string",
|
|
"coords": "string",
|
|
"topic": "string",
|
|
}
|
|
)
|
|
domain_profile: str = "generic_news"
|
|
extract_mode: Literal["heuristic", "llm"] = "heuristic"
|
|
instruction: str | None = None
|
|
|
|
@field_validator("urls")
|
|
@classmethod
|
|
def non_empty_urls(cls, value: list[str]) -> list[str]:
|
|
cleaned = [u.strip() for u in value if u and u.strip()]
|
|
if not cleaned:
|
|
raise ValueError("urls must contain at least one URL")
|
|
return cleaned
|
|
|
|
|
|
class ViinaSourceConfig(BaseModel):
|
|
urls: list[str] = Field(default_factory=list)
|
|
texts: list[str] = Field(default_factory=list)
|
|
input_mode: Literal["urls", "texts", "mixed"] = "urls"
|
|
|
|
@field_validator("urls")
|
|
@classmethod
|
|
def strip_urls(cls, value: list[str]) -> list[str]:
|
|
return [u.strip() for u in value if u and u.strip()]
|
|
|
|
@field_validator("texts")
|
|
@classmethod
|
|
def strip_texts(cls, value: list[str]) -> list[str]:
|
|
return [t.strip() for t in value if t and t.strip()]
|
|
|
|
@model_validator(mode="after")
|
|
def require_inputs(self) -> "ViinaSourceConfig":
|
|
if self.input_mode == "urls" and not self.urls:
|
|
raise ValueError("urls required when input_mode=urls")
|
|
if self.input_mode == "texts" and not self.texts:
|
|
raise ValueError("texts required when input_mode=texts")
|
|
if self.input_mode == "mixed" and not self.urls and not self.texts:
|
|
raise ValueError("urls or texts required when input_mode=mixed")
|
|
return self
|
|
|
|
|
|
CONFIG_MODELS: dict[str, type[BaseModel]] = {
|
|
"telegram": TelegramSourceConfig,
|
|
"crawl4ai": Crawl4AISourceConfig,
|
|
"viina": ViinaSourceConfig,
|
|
}
|
|
|
|
|
|
def parse_source_config(source_type: str, raw: dict) -> BaseModel:
|
|
model = CONFIG_MODELS.get(source_type)
|
|
if model is None:
|
|
raise ValueError(f"Unknown source_type: {source_type}")
|
|
return model.model_validate(raw or {})
|