Add CP→CA→PI platform foundation on MapMil monorepo.

Unify parsing workers, analytics API with PostgreSQL, map UI, and PI distribution into centers/ with Docker Compose.

Co-authored-by: Cursor <cursoragent@cursor.com>

centers/parsing/workers/worker.py

@@ -0,0 +1,132 @@
+"""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()