103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
import os
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
from telethon import TelegramClient
|
|
from db.repository import Repository
|
|
from core.metrics import POSTS_PUBLISHED_TOTAL, QUEUE_POSTS_GAUGE
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class PublisherService:
|
|
def __init__(
|
|
self,
|
|
repo: Repository,
|
|
api_id: Optional[int] = None,
|
|
api_hash: Optional[str] = None,
|
|
session_name: str = "/projects/telegram-bots/copykar/sessions/publisher.session",
|
|
bot_token: Optional[str] = None,
|
|
):
|
|
self.repo = repo
|
|
self.api_id = api_id or int(os.getenv("API_ID", "0"))
|
|
self.api_hash = api_hash or os.getenv("API_HASH", "")
|
|
self.bot_token = bot_token or os.getenv("BOT_TOKEN")
|
|
self.session_name = session_name
|
|
self.client = TelegramClient(self.session_name, self.api_id, self.api_hash)
|
|
self._running = False
|
|
self._task: Optional[asyncio.Task] = None
|
|
|
|
async def start(self):
|
|
os.makedirs(os.path.dirname(self.session_name), exist_ok=True)
|
|
logger.info("Starting Publisher Service...")
|
|
if self.bot_token:
|
|
await self.client.start(bot_token=self.bot_token)
|
|
else:
|
|
await self.client.start()
|
|
logger.info("Publisher Service connected successfully.")
|
|
|
|
self._running = True
|
|
self._task = asyncio.create_task(self._publisher_loop())
|
|
|
|
async def _publisher_loop(self):
|
|
while self._running:
|
|
try:
|
|
await self._process_pending_queues()
|
|
except Exception as e:
|
|
logger.error(f"Error in publisher loop: {e}", exc_info=True)
|
|
await asyncio.sleep(15) # Check queue every 15 seconds
|
|
|
|
async def _process_pending_queues(self):
|
|
targets = await self.repo.get_active_targets()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# Update gauge metrics
|
|
pending_count = len(await self.repo.get_posts_by_status("approved", limit=5000))
|
|
QUEUE_POSTS_GAUGE.labels(status="approved").set(pending_count)
|
|
|
|
for target in targets:
|
|
# Check interval cooldown
|
|
if target.last_post_time:
|
|
last_post = target.last_post_time
|
|
if last_post.tzinfo is None:
|
|
last_post = last_post.replace(tzinfo=timezone.utc)
|
|
diff_minutes = (now - last_post).total_seconds() / 60.0
|
|
if diff_minutes < target.post_interval_min:
|
|
continue # Cooldown not elapsed yet
|
|
|
|
# Fetch next approved post for this target
|
|
post = await self.repo.get_next_approved_post_for_target(target.id)
|
|
if not post:
|
|
continue
|
|
|
|
try:
|
|
publish_text = post.ai_text or post.raw_text or ""
|
|
if post.media_path and os.path.exists(post.media_path):
|
|
await self.client.send_file(
|
|
target.channel_id,
|
|
file=post.media_path,
|
|
caption=publish_text,
|
|
parse_mode="markdown"
|
|
)
|
|
else:
|
|
await self.client.send_message(
|
|
target.channel_id,
|
|
publish_text,
|
|
parse_mode="markdown"
|
|
)
|
|
|
|
await self.repo.mark_post_published(post.id)
|
|
await self.repo.update_target_last_post(target.id)
|
|
POSTS_PUBLISHED_TOTAL.labels(target_channel_id=str(target.channel_id)).inc()
|
|
logger.info(f"Successfully published post {post.id} to target channel {target.title} ({target.channel_id})")
|
|
except Exception as e:
|
|
logger.error(f"Failed to publish post {post.id} to target {target.channel_id}: {e}", exc_info=True)
|
|
|
|
async def stop(self):
|
|
self._running = False
|
|
if self._task:
|
|
self._task.cancel()
|
|
if self.client.is_connected():
|
|
await self.client.disconnect()
|
|
logger.info("Publisher Service disconnected.")
|