gitrusprusandCursor
9dfe713668
Unify parsing workers, analytics API with PostgreSQL, map UI, and PI distribution into centers/ with Docker Compose. Co-authored-by: Cursor <cursoragent@cursor.com>
133 lines
4.1 KiB
Python
133 lines
4.1 KiB
Python
"""CP worker: poll Redis jobs, parse Telegram, ingest to CA."""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
|
|
import httpx
|
|
import redis
|
|
|
|
from workers.converter import event_record_to_ingest
|
|
from workers.parsers.telegram_events import parse_event_posts
|
|
from workers.sources.telegram_client import (
|
|
TelegramAuthError,
|
|
TelegramConfigError,
|
|
fetch_channel_posts,
|
|
normalize_channel,
|
|
)
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger("cp-worker")
|
|
|
|
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
|
|
JOB_QUEUE_KEY = "cp:jobs"
|
|
CA_API_URL = os.getenv("CA_API_URL", "http://ca-api:8000")
|
|
INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "dev-internal-token")
|
|
POLL_TIMEOUT = int(os.getenv("WORKER_POLL_TIMEOUT", "5"))
|
|
|
|
|
|
def get_redis() -> redis.Redis:
|
|
return redis.from_url(REDIS_URL, decode_responses=True)
|
|
|
|
|
|
async def process_telegram_job(job_id: int, source_config: dict) -> tuple[list[dict], str | None]:
|
|
channel = source_config.get("channel", "creamy_caprice")
|
|
limit = int(source_config.get("limit", 100))
|
|
|
|
try:
|
|
username = normalize_channel(channel)
|
|
posts = await fetch_channel_posts(username, limit=limit)
|
|
except (TelegramConfigError, TelegramAuthError, ValueError) as exc:
|
|
return [], str(exc)
|
|
except Exception as exc:
|
|
return [], f"Telegram: {exc}"
|
|
|
|
records = parse_event_posts(posts)
|
|
events = [event_record_to_ingest(r) for r in records]
|
|
return events, None
|
|
|
|
|
|
async def post_ingest(job_id: int, events: list[dict]) -> None:
|
|
if not events:
|
|
return
|
|
|
|
async with httpx.AsyncClient(timeout=120.0) as client:
|
|
response = await client.post(
|
|
f"{CA_API_URL}/internal/ingest",
|
|
json={"job_id": job_id, "events": events},
|
|
headers={"X-Internal-Token": INTERNAL_TOKEN},
|
|
)
|
|
response.raise_for_status()
|
|
logger.info("Ingested %s events for job %s: %s", len(events), job_id, response.json())
|
|
|
|
|
|
async def patch_job_status(job_id: int, status: str, error: str | None = None) -> None:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
params = {"status": status}
|
|
if error:
|
|
params["error"] = error
|
|
response = await client.patch(
|
|
f"{CA_API_URL}/internal/jobs/{job_id}",
|
|
params=params,
|
|
headers={"X-Internal-Token": INTERNAL_TOKEN},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
|
|
async def handle_job(payload: dict) -> None:
|
|
job_id = payload["job_id"]
|
|
source_type = payload["source_type"]
|
|
source_config = payload.get("source_config", {})
|
|
|
|
logger.info("Processing job %s (%s)", job_id, source_type)
|
|
await patch_job_status(job_id, "running")
|
|
|
|
if source_type == "telegram":
|
|
events, error = await process_telegram_job(job_id, source_config)
|
|
else:
|
|
events, error = [], f"Unsupported source_type: {source_type}"
|
|
|
|
if error:
|
|
logger.error("Job %s failed: %s", job_id, error)
|
|
await patch_job_status(job_id, "failed", error=error)
|
|
return
|
|
|
|
try:
|
|
await post_ingest(job_id, events)
|
|
if not events:
|
|
await patch_job_status(job_id, "completed", error="No events found")
|
|
logger.info("Job %s completed with %s events", job_id, len(events))
|
|
except Exception as exc:
|
|
logger.exception("Ingest failed for job %s", job_id)
|
|
await patch_job_status(job_id, "failed", error=str(exc))
|
|
|
|
|
|
async def worker_loop() -> None:
|
|
r = get_redis()
|
|
logger.info("CP worker started, polling %s", JOB_QUEUE_KEY)
|
|
|
|
while True:
|
|
try:
|
|
item = r.blpop(JOB_QUEUE_KEY, timeout=POLL_TIMEOUT)
|
|
if not item:
|
|
continue
|
|
_, raw = item
|
|
payload = json.loads(raw)
|
|
await handle_job(payload)
|
|
except redis.RedisError as exc:
|
|
logger.error("Redis error: %s", exc)
|
|
time.sleep(3)
|
|
except Exception:
|
|
logger.exception("Unexpected worker error")
|
|
time.sleep(1)
|
|
|
|
|
|
def main() -> None:
|
|
asyncio.run(worker_loop())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|