"""Custom JSON encoder that handles datetime objects."""

def default(self, obj):

if isinstance(obj, datetime):

return obj.isoformat()

if isinstance(obj, bytes):

return obj.decode("utf-8", errors="ignore")

"""Load configuration from config.toml"""

config_path = Path(__file__).parent / "config.toml"

if not config_path.exists():

raise FileNotFoundError(f"Config file not found: {config_path}")

# Load TOML file (always use binary mode for tomllib/tomli)

with open(config_path, "rb") as f:

"""Save checkpoint data to resume processing if interrupted."""

checkpoint_dir = Path("raw/checkpoints")

checkpoint_dir.mkdir(parents=True, exist_ok=True)

checkpoint_file = checkpoint_dir / f"{channel}_checkpoint.json"

output_config = config.get("output", {})

indent_spaces = output_config.get("indent_spaces", 2)

checkpoint_data = {

"channel": channel,

"timestamp": datetime.now(timezone.utc).isoformat(),

"message_count": len(messages),

"last_message_id": messages[-1]["id"] if messages else None,

"messages": messages,

"user_info": user_info,

}

with open(checkpoint_file, "w", encoding="utf-8") as f:

json.dump(checkpoint_data, f, indent=indent_spaces, ensure_ascii=False)

"""Load checkpoint data if it exists."""

checkpoint_file = Path("raw/checkpoints") / f"{channel}_checkpoint.json"

if checkpoint_file.exists():

with open(checkpoint_file, "r", encoding="utf-8") as f:

return json.load(f)

"""Remove checkpoint file after successful completion."""

checkpoint_file = Path("raw/checkpoints") / f"{channel}_checkpoint.json"

if checkpoint_file.exists():

checkpoint_file.unlink()

"""Fetch messages from a Telegram channel within the configured time window."""

crawler_config = config.get("crawler", {})

group_chats_config = config.get("group_chats", {})

# Check if channel is excluded

excluded_channels = group_chats_config.get("exclude", [])

if channel in excluded_channels:

print(f"⏭️ Skipping excluded channel: {channel}")

return {"channel": channel, "messages": [], "skipped": True}

time_window_days = crawler_config.get("time_window_days", 3)

max_messages = crawler_config.get("max_messages_per_channel", 2000)

rate_delay = crawler_config.get("rate_limiting_delay", 0.5)

batch_size = crawler_config.get(

"batch_size", 500

) # Fetch messages in batches (default 500)

checkpoint_interval = crawler_config.get(

"checkpoint_interval", 100

) # Save every N messages

offset_date = datetime.now(timezone.utc) - timedelta(days=time_window_days)

# Check for existing checkpoint

checkpoint = load_checkpoint(channel)

if checkpoint:

print(

f"📂 Found checkpoint for channel {channel} with {checkpoint['message_count']} messages"

)

print(f" Last saved: {checkpoint['timestamp']}")

resume = input(" Resume from checkpoint? (y/n): ").strip().lower()

if resume == "y":

messages_with_reactions = checkpoint["messages"]

user_info = checkpoint["user_info"]

offset_id = checkpoint["last_message_id"]

total_fetched = len(messages_with_reactions)

print(f"✅ Resuming from message ID {offset_id}")

else:

messages_with_reactions = []

user_info = {}

offset_id = 0

total_fetched = 0

clear_checkpoint(channel)

else:

messages_with_reactions = []

user_info = {} # Track user_id -> {username, first_name, last_name, photo_url, bio}

offset_id = 0

total_fetched = 0

try:

print(f"📥 Fetching messages from {channel} in batches of {batch_size}...")

print(f" ⏰ Time window: fetching messages since {offset_date.isoformat()}")

if offset_id > 0:

batch_number = (total_fetched // batch_size) + 1

else:

batch_number = 1

while total_fetched < max_messages:

# Calculate how many messages to fetch in this batch

messages_to_fetch = min(batch_size, max_messages - total_fetched)

# Fetch a batch of messages (starting from most recent, going backwards)

batch_messages = await client.get_messages(

channel,

limit=messages_to_fetch,

offset_id=offset_id,

)

if not batch_messages:

# No more messages to fetch

break

print(f" 📦 Batch {batch_number}: Fetched {len(batch_messages)} messages")

if batch_messages:

first_msg_date = (

batch_messages[0].date.isoformat()

if batch_messages[0].date

else "None"

)

last_msg_date = (

batch_messages[-1].date.isoformat()

if batch_messages[-1].date

else "None"

)

print(f" First message date: {first_msg_date}")

print(f" Last message date: {last_msg_date}")

batch_processed_count = 0

reached_time_limit = False

for idx, message in enumerate(batch_messages, 1):

# Check if message is within time window

if message.date and message.date < offset_date:

# We've gone past our time window, stop fetching

print(

f" ⏹️ Reached end of time window at message {message.id} (date: {message.date.isoformat()})"

)

print(

f" Processed {batch_processed_count} messages from this batch before time limit"

)

reached_time_limit = True

break

batch_processed_count += 1

# Collect user info from message sender

if message.from_id and hasattr(message.from_id, "user_id"):

user_id = message.from_id.user_id

if user_id not in user_info:

try:

user = await client.get_entity(user_id)

# Get profile photo URL if available

photo_url = ""

if user.photo:

try:

# Download profile photo to get the file reference

photo = await client.download_profile_photo(

user_id, file=bytes

)

if photo:

# Construct a URL-like identifier for the photo

# Note: Telegram doesn't provide direct URLs, so we store photo ID

photo_url = (

f"photo:{user.photo.photo_id}"

if hasattr(user.photo, "photo_id")

else ""

)

except Exception:

pass

# Get bio/description

bio = ""

try:

full_user = await client(GetFullUserRequest(user_id))

if full_user.full_user.about:

bio = full_user.full_user.about

except Exception:

pass

user_info[user_id] = {

"username": user.username or "",

"first_name": user.first_name or "",

"last_name": user.last_name or "",

"photo_url": photo_url,

"bio": bio,

}

except Exception:

# If we can't get user info, just skip it

pass

# Show progress for reactions fetching

if idx % 50 == 0 or idx == len(batch_messages):

print(

f" Processing reactions: {idx}/{len(batch_messages)} messages..."

)

# Fetch detailed reactions with user IDs for this message

reaction_details = []

if message.reactions and message.reactions.results:

try:

# Get message reactions with user details using GetMessageReactionsList

from telethon.tl.functions.messages import (

GetMessageReactionsListRequest,

)

for reaction_result in message.reactions.results:

emoji = (

reaction_result.reaction.emoticon

if hasattr(reaction_result.reaction, "emoticon")

else None

)

if emoji:

# Get users who reacted with this specific emoji

result = await client(

GetMessageReactionsListRequest(

peer=channel,

id=message.id,

reaction=reaction_result.reaction,

limit=100,

)

)

# Extract user IDs from the reaction list

for reaction_peer in result.reactions:

user_id = (

reaction_peer.peer_id.user_id

if hasattr(reaction_peer.peer_id, "user_id")

else None

)

if user_id:

reaction_details.append(

{"user_id": user_id, "emoji": emoji}

)

# Collect user info from reactions

if user_id not in user_info:

try:

user = await client.get_entity(user_id)

# Get profile photo URL if available

photo_url = ""

if user.photo:

try:

photo_url = (

f"photo:{user.photo.photo_id}"

if hasattr(

user.photo, "photo_id"

)

else ""

)

except Exception:

pass

# Get bio/description

bio = ""

try:

full_user = await client(

GetFullUserRequest(user_id)

)

if full_user.full_user.about:

bio = full_user.full_user.about

except Exception:

pass

user_info[user_id] = {

"username": user.username or "",

"first_name": user.first_name or "",

"last_name": user.last_name or "",

"photo_url": photo_url,

"bio": bio,

}

except Exception:

pass

except Exception as e:

# Fallback to basic reaction without user IDs if detailed fetch fails

for reaction_result in message.reactions.results:

emoji = (

reaction_result.reaction.emoticon

if hasattr(reaction_result.reaction, "emoticon")

else None

)

if emoji:

reaction_details.append(

{

"user_id": None,

"emoji": emoji,

"count": reaction_result.count,

}

)

# Convert message to simplified format immediately for checkpoint

simplified_msg = {

"id": message.id,

"date": message.date.isoformat() if message.date else None,

"from_id": message.from_id.user_id

if hasattr(message.from_id, "user_id")

else None,

"message": message.message,

"reply_to_msg_id": message.reply_to.reply_to_msg_id

if message.reply_to

else None,

"reactions": reaction_details if reaction_details else [],

"replies": message.replies.replies if message.replies else None,

}

messages_with_reactions.append(simplified_msg)

total_fetched += batch_processed_count

# Save checkpoint periodically

if checkpoint_interval > 0 and total_fetched % checkpoint_interval == 0:

save_checkpoint(channel, messages_with_reactions, user_info, config)

# If we reached the time limit, stop fetching more batches

if reached_time_limit:

break

# Rate limiting delay between batches

await asyncio.sleep(rate_delay)

# Update offset_id to the ID of the last (oldest) message in this batch

offset_id = batch_messages[-1].id

batch_number += 1

# If we got fewer messages than requested, we've reached the end

if len(batch_messages) < messages_to_fetch:

break

print(

f"✅ Fetched {len(messages_with_reactions)} messages from {channel} in {batch_number} batches"

)

print(f"👥 Collected info for {len(user_info)} unique users")

# Clear checkpoint after successful completion

clear_checkpoint(channel)

# Add post counts to user_info

return {

"channel": channel,

"messages": messages_with_reactions,

"user_info": user_info,

"skipped": False,

}

except Exception as e:

print(f"❌ Error fetching messages from {channel}: {e}")

"""Save fetched messages to JSON files."""

output_config = config.get("output", {})

pretty_print = output_config.get("pretty_print", True)

indent_spaces = output_config.get("indent_spaces", 2)

# Create raw directory if it doesn't exist

output_dir = Path("raw")

output_dir.mkdir(parents=True, exist_ok=True)

for result in results:

# Skip excluded or errored channels

if result.get("skipped", False) or result.get("error"):

continue

# Save user info to CSV if available

user_info = result.get("user_info", {})

if user_info:

csv_filename = output_dir / f"{result['channel']}_user_ids.csv"

with open(csv_filename, "w", encoding="utf-8") as f:

f.write("user_id,username,first_name,last_name,photo_url,bio\n")

for user_id, info in sorted(user_info.items()):

username = info.get("username", "")

first_name = info.get("first_name", "").replace(",", " ")

last_name = info.get("last_name", "").replace(",", " ")

photo_url = info.get("photo_url", "")

bio = info.get("bio", "").replace(",", " ").replace("\n", " ")

f.write(

f"{user_id},{username},{first_name},{last_name},{photo_url},{bio}\n"

)

print(f"👥 Saved {len(user_info)} users to {csv_filename}")

# Messages are already in simplified format from fetch_channel_messages

serializable_messages = result["messages"]

# Save to raw directory with _messages.json suffix

filename = output_dir / f"{result['channel']}_messages.json"

indent = indent_spaces if pretty_print else None

with open(filename, "w", encoding="utf-8") as f:

json.dump(

serializable_messages,

f,

indent=indent,

ensure_ascii=False,

)

"""Main entry point for the Telegram crawler."""

print("🚀 Starting Telegram Channel Crawler\n")

# Load configuration

config = load_config()

crawler_config = config.get("crawler", {})

group_chats_config = config.get("group_chats", {})

# Get environment variables

api_id = int(os.getenv("TELEGRAM_APP_ID", "0"))

api_hash = os.getenv("TELEGRAM_APP_HASH", "")

phone = os.getenv("TELEGRAM_PHONE", "")

if not api_id or not api_hash:

print("❌ Error: Missing Telegram credentials in .env file")

print("Please set TELEGRAM_APP_ID and TELEGRAM_APP_HASH")

print("\nGet your credentials from: https://my.telegram.org")

sys.exit(1)

# Get group chats to crawl (must be integers)

channels = group_chats_config.get("include", [])

if not channels:

print("❌ Error: No group chats configured in config.toml")

sys.exit(1)

# Validate that all channels are integers

for ch in channels:

if not isinstance(ch, int):

print(f"❌ Error: Channel '{ch}' is not a valid ID (must be an integer)")

print("Use 'python list_channels.py' to see your channel IDs")

sys.exit(1)

parallel_requests = crawler_config.get("parallel_requests", 3)

# Initialize Telegram client with session file

session_file = "telegram_session"

client = TelegramClient(session_file, api_id, api_hash)

try:

print("🔌 Connecting to Telegram...")

await client.start(

phone=lambda: phone

or input(

"📱 Enter your phone number (with country code, e.g., +1234567890): "

),

password=lambda: input("🔒 Enter your 2FA password (if enabled): "),

code_callback=lambda: input("💬 Enter the code Telegram sent you: "),

)

if not await client.is_user_authorized():

print("❌ Authorization failed. Please try again.")

sys.exit(1)

me = await client.get_me()

print(f"✅ Connected as: {me.first_name} (@{me.username or 'no username'})\n")

# Process channels in parallel batches

all_results = []

for chunk in chunk_array(channels, parallel_requests):

tasks = [

fetch_channel_messages(client, channel, config) for channel in chunk

]

results = await asyncio.gather(*tasks, return_exceptions=True)

# Separate successful results from errors

for result in results:

if isinstance(result, Exception):

print(f"❌ Error: {result}")

else:

all_results.append(result)

# Save all messages

print()

await save_messages(all_results, config)

print("\n✅ Crawling complete!")

except Exception as e:

print(f"❌ Fatal error: {e}")

sys.exit(1)

finally:

if client.is_connected():

await client.disconnect()