saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
from app.models.enums import MonitoringFrequency, SourceStatus, SourceType
|
|
|
|
# Only these types accept a user-supplied URL via the API. Every other type
|
|
# (website/github/sec_edgar/job_posting) is populated by discovery instead,
|
|
# and patent/review only ever activate through a configured fixture key.
|
|
_USER_CREATABLE_TYPES = {SourceType.CUSTOM_URL, SourceType.RSS}
|
|
|
|
|
|
class SourceCreate(BaseModel):
|
|
source_type: SourceType
|
|
name: str = Field(min_length=1, max_length=200)
|
|
base_url: str = Field(min_length=1, max_length=500)
|
|
|
|
@field_validator("source_type")
|
|
@classmethod
|
|
def _validate_type(cls, value: SourceType) -> SourceType:
|
|
if value not in _USER_CREATABLE_TYPES:
|
|
raise ValueError(
|
|
f"{value.value} sources are created by discovery, not added directly. "
|
|
f"Only {', '.join(t.value for t in _USER_CREATABLE_TYPES)} may be added here."
|
|
)
|
|
return value
|
|
|
|
@field_validator("base_url")
|
|
@classmethod
|
|
def _normalize_url(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not (value.startswith("http://") or value.startswith("https://")):
|
|
value = f"https://{value}"
|
|
return value
|
|
|
|
|
|
class SourceUpdate(BaseModel):
|
|
name: str | None = Field(default=None, min_length=1, max_length=200)
|
|
active: bool | None = None
|
|
# None means "inherit the company's default cadence" - the default for
|
|
# every source. Sending frequency_type: null explicitly clears an
|
|
# existing override back to that default.
|
|
frequency_type: MonitoringFrequency | None = None
|
|
interval_minutes: int | None = Field(default=None, ge=1)
|
|
cron_expression: str | None = Field(default=None, max_length=120)
|
|
|
|
|
|
class SourceResponse(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: uuid.UUID
|
|
source_type: SourceType
|
|
name: str
|
|
base_url: str | None
|
|
active: bool
|
|
status: SourceStatus
|
|
trust_score: float
|
|
last_checked: datetime | None
|
|
last_successful_check: datetime | None
|
|
failure_count: int
|
|
frequency_type: MonitoringFrequency | None
|
|
interval_minutes: int | None
|
|
cron_expression: str | None
|
|
next_check: datetime | None
|
|
|
|
|
|
class SourceTestResult(BaseModel):
|
|
status: SourceStatus
|
|
documents_found: int
|
|
error: str | None
|